import org.threadly.util.Clock; //导入依赖的package包/类
@Test
public void setMaxScriptStepRateTest() throws InterruptedException, ExecutionException {
  int testSteps = 5;
  SequentialScriptBuilder builder = new SequentialScriptBuilder();
  builder.setMaxScriptStepRate(testSteps * 10);
  for (int i = 0; i < testSteps + 1; i++) {
    builder.addStep(new TestStep());
  }
  ExecutableScript script = builder.build();
  
  long start = Clock.accurateForwardProgressingMillis();
  FutureUtils.blockTillAllCompleteOrFirstError(script.startScript());
  long end = Clock.accurateForwardProgressingMillis();
  
  assertTrue(end - start >= 100);
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
@Override
public void run() {
  long startTime = threadRunTime > 0 ? Clock.accurateForwardProgressingMillis() : -1;
  if (run) {
    countArray.incrementAndGet(index);
    
    long scheduleDelay;
    if (DIFFER_SCHEDULE_TIME) {
      scheduleDelay = AbstractSchedulerScheduleBenchmark.this.scheduleDelay * (index / 2);
    } else {
      scheduleDelay = AbstractSchedulerScheduleBenchmark.this.scheduleDelay;
    }
    
    doThreadWork(startTime);
    
    getScheduler().schedule(this, scheduleDelay);
  }
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
@Override
protected OneTimeTaskWrapper doSchedule(Runnable task, long delayInMillis, TaskPriority priority) {
  QueueSet queueSet = taskQueueManager.getQueueSet(priority);
  OneTimeTaskWrapper result;
  if (delayInMillis == 0) {
    addToExecuteQueue(queueSet, 
                      (result = new OneTimeTaskWrapper(task, queueSet.executeQueue, 
                                                       Clock.lastKnownForwardProgressingMillis())));
  } else {
    addToScheduleQueue(queueSet, 
                       (result = new OneTimeTaskWrapper(task, queueSet.scheduleQueue, 
                                                        Clock.accurateForwardProgressingMillis() + 
                                                          delayInMillis)));
  }
  return result;
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
@Override
public void scheduleAtFixedRate(Runnable task, long initialDelay, long period, 
                                TaskPriority priority) {
  ArgumentVerifier.assertNotNull(task, "task");
  ArgumentVerifier.assertNotNegative(initialDelay, "initialDelay");
  ArgumentVerifier.assertGreaterThanZero(period, "period");
  if (priority == null) {
    priority = defaultPriority;
  }

  QueueSet queueSet = taskQueueManager.getQueueSet(priority);
  addToScheduleQueue(queueSet, 
                     new RecurringRateTaskWrapper(task, queueSet, 
                                                  Clock.accurateForwardProgressingMillis() + initialDelay, 
                                                  period));
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
/**
 * Call to calculate how many milliseconds until the provided time.  If we are past the 
 * provided hour/minute, it will be the milliseconds until we reach that time with the NEXT day.  
 * <p>
 * It is important to note that the time zone for this hour is UTC.  If you want to use this for 
 * local time, just pass the hour through {@link #shiftLocalHourToUTC(int)}.  This will convert 
 * a local time's hour to UTC so that it can be used in this invocation.  
 * <p>
 * Because of use of {@link Clock#lastKnownTimeMillis()}, this calculation will only be accurate 
 * within about 100 milliseconds.  Of course if provided to a scheduler, depending on it's work 
 * load that variation may be greater.
 * 
 * @param now Current time in milliseconds since epoc
 * @param hour Hour in the 24 hour format, can not be negative and must be less than 24
 * @param minute Minute to calculate too, can not be negative and must be less than 60
 * @return Time in milliseconds till provided time is reached
 */
protected static long getDelayTillHour(long now, int hour, int minute) {
  long delayInMillis = TimeUnit.MINUTES.toMillis(minute);
  long currentHour = TimeUnit.MILLISECONDS.toHours(now % TimeUnit.DAYS.toMillis(1));
  
  if (hour > currentHour) {
    delayInMillis += TimeUnit.HOURS.toMillis(hour - currentHour);
  } else if (hour < currentHour) {
    delayInMillis += TimeUnit.HOURS.toMillis(TimeUnit.DAYS.toHours(1) - currentHour + hour);
  } else {
    long result = getDelayTillMinute(Clock.lastKnownTimeMillis(), minute);
    if (TimeUnit.MILLISECONDS.toMinutes(result) <= minute) {
      return result;
    } else {
      // here we have to add the time to forward us to the next day
      return result + TimeUnit.HOURS.toMillis(TimeUnit.DAYS.toHours(1) - 1);
    }
  }

  // subtract minutes, seconds, and milliseconds that have passed
  long offset = now % TimeUnit.HOURS.toMillis(1);
  return delayInMillis - offset;
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
/**
 * This call blocks till all futures in the list have completed.  If the future completed with 
 * an error an {@link ExecutionException} is thrown.  If this exception is thrown, all futures 
 * may or may not be completed, the exception is thrown as soon as it is hit.  There also may be 
 * additional futures that errored (but were not hit yet).
 * 
 * @since 4.0.0
 * @param futures Structure of futures to iterate over
 * @param timeoutInMillis timeout to wait for futures to complete in milliseconds
 * @throws InterruptedException Thrown if thread is interrupted while waiting on future
 * @throws TimeoutException Thrown if the timeout elapsed while waiting on futures to complete
 * @throws ExecutionException Thrown if future throws exception on .get() call
 */
public static void blockTillAllCompleteOrFirstError(Iterable<? extends Future<?>> futures, long timeoutInMillis) 
    throws InterruptedException, TimeoutException, ExecutionException {
  if (futures == null) {
    return;
  }
  
  Iterator<? extends Future<?>> it = futures.iterator();
  long startTime = Clock.accurateForwardProgressingMillis();
  long remainingTime;
  while (it.hasNext() && 
         (remainingTime = timeoutInMillis - (Clock.lastKnownForwardProgressingMillis() - startTime)) > 0) {
    it.next().get(remainingTime, TimeUnit.MILLISECONDS);
  }
  if (it.hasNext()) {
    throw new TimeoutException();
  }
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
@Override
public T get(long timeout, TimeUnit unit) throws InterruptedException, 
                                                 ExecutionException, TimeoutException {
  long startTime = Clock.accurateForwardProgressingMillis();
  long timeoutInMs = unit.toMillis(timeout);
  synchronized (resultLock) {
    long remainingInMs;
    while (! done && 
           (remainingInMs = timeoutInMs - (Clock.accurateForwardProgressingMillis() - startTime)) > 0) {
      resultLock.wait(remainingInMs);
    }
    if (resultCleared) {
      throw new IllegalStateException("Result cleared, future get's not possible");
    }
    
    if (failure != null) {
      throw new ExecutionException(failure);
    } else if (canceled) {
      throw new CancellationException();
    } else if (done) {
      return result;
    } else {
      throw new TimeoutException();
    }
  }
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
public void trackStart(Pair<Thread, Runnable> taskPair, long expectedRunTime) {
  // get start time before any operations for hopefully more accurate execution delay
  long startTime = Clock.accurateForwardProgressingMillis();
  
  queuedTaskCount.decrement();
  totalExecutionCount.increment();
  
  synchronized (runDelays) {
    runDelays.addLast(startTime - expectedRunTime);
    if (runDelays.size() > maxStatisticWindowSize) {
      runDelays.removeFirst();
    }
  }
  
  // get possibly newer time so we don't penalize stats tracking as duration
  runningTasks.put(taskPair, Clock.lastKnownForwardProgressingMillis());
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
/**
 * Constructs a new {@link RateLimiterExecutor}.  Tasks will be scheduled on the provided 
 * scheduler, so it is assumed that the scheduler will have enough threads to handle the 
 * average permit amount per task, per second.  
 * <p>
 * This constructor accepts a maximum schedule delay.  If a task requires being scheduled out 
 * beyond this delay, then the provided {@link RejectedExecutionHandler} will be invoked.
 * 
 * @since 4.8.0
 * @param scheduler scheduler to schedule/execute tasks on
 * @param permitsPerSecond how many permits should be allowed per second
 * @param maxScheduleDelayMillis Maximum amount of time delay tasks in order to maintain rate
 * @param rejectedExecutionHandler Handler to accept tasks which could not be executed
 */
public RateLimiterExecutor(SubmitterScheduler scheduler, double permitsPerSecond, 
                           long maxScheduleDelayMillis, 
                           RejectedExecutionHandler rejectedExecutionHandler) {
  ArgumentVerifier.assertNotNull(scheduler, "scheduler");
  
  this.scheduler = scheduler;
  if (rejectedExecutionHandler == null) {
    rejectedExecutionHandler = RejectedExecutionHandler.THROW_REJECTED_EXECUTION_EXCEPTION;
  }
  this.rejectedExecutionHandler = rejectedExecutionHandler;
  this.permitLock = new Object();
  this.lastScheduleTime = Clock.lastKnownForwardProgressingMillis();
  setPermitsPerSecond(permitsPerSecond);
  setMaxScheduleDelayMillis(maxScheduleDelayMillis);
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
@Test
public void blockTillRunTest() {
  TestRunnable tr = new TestRunnable() {
    private boolean firstRun = true;
    
    @Override
    public void handleRunStart() throws InterruptedException {
      if (firstRun) {
        firstRun = false;
        TestUtils.sleep(DELAY_TIME);
        run();
      }
    }
  };
  new Thread(tr).start();
  
  long startTime = Clock.accurateForwardProgressingMillis();
  tr.blockTillFinished(1000, 2);
  long endTime = Clock.accurateForwardProgressingMillis();
  
  assertTrue(endTime - startTime >= (DELAY_TIME - ALLOWED_VARIANCE));
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
public List<Pair<Runnable, StackTraceElement[]>> getLongRunningTasks(long durationLimitMillis) {
  List<Pair<Runnable, StackTraceElement[]>> result = new ArrayList<>();
  if (accurateTime) {
    // ensure clock is updated before loop
    Clock.accurateForwardProgressingMillis();
  }
  for (Map.Entry<Pair<Thread, TaskStatWrapper>, Long> e : runningTasks.entrySet()) {
    if (Clock.lastKnownForwardProgressingMillis() - e.getValue() > durationLimitMillis) {
      Runnable task = e.getKey().getRight().task;
      if (task instanceof ListenableFutureTask) {
        ListenableFutureTask<?> lft = (ListenableFutureTask<?>)task;
        if (lft.getContainedCallable() instanceof RunnableCallableAdapter) {
          RunnableCallableAdapter<?> rca = (RunnableCallableAdapter<?>)lft.getContainedCallable();
          task = rca.getContainedRunnable();
        }
      }
      StackTraceElement[] stack = e.getKey().getLeft().getStackTrace();
      // verify still in collection after capturing stack
      if (runningTasks.containsKey(e.getKey())) {
        result.add(new Pair<>(task, stack));
      }
    }
  }
  
  return result;
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
public int getLongRunningTasksQty(long durationLimitMillis) {
  int result = 0;
  
  long now = accurateTime ? 
               Clock.accurateForwardProgressingMillis() : 
               Clock.lastKnownForwardProgressingMillis();
  Iterator<Long> it = runningTasks.values().iterator();
  while (it.hasNext()) {
    Long startTime = it.next();
    if (now - startTime >= durationLimitMillis) {
      result++;
    }
  }
  
  return result;
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
@Override
public TaskWrapper workerIdle(Worker worker) {
  TaskWrapper result = super.workerIdle(worker);

  // may not be a wrapper for internal tasks like shutdown
  if (result != null && result.getContainedRunnable() instanceof TaskStatWrapper) {
    long taskDelay = Clock.lastKnownForwardProgressingMillis() - result.getPureRunTime();
    TaskStatWrapper statWrapper = (TaskStatWrapper)result.getContainedRunnable();
    ConcurrentArrayList<Long> priorityStats = 
        statsManager.getExecutionDelaySamplesInternal(statWrapper.priority);
  
    synchronized (priorityStats.getModificationLock()) {
      priorityStats.add(taskDelay);
      statsManager.trimWindow(priorityStats);
    }
  }
  
  return result;
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
@Override
protected TaskWrapper getNextReadyTask() {
  TaskWrapper result = super.getNextReadyTask();
  
  // may not be a wrapper for internal tasks like shutdown
  if (result != null && result.getContainedRunnable() instanceof TaskStatWrapper) {
    long taskDelay = Clock.lastKnownForwardProgressingMillis() - result.getPureRunTime();
    TaskStatWrapper statWrapper = (TaskStatWrapper)result.getContainedRunnable();
    ConcurrentArrayList<Long> priorityStats = 
        statsManager.getExecutionDelaySamplesInternal(statWrapper.priority);

    synchronized (priorityStats.getModificationLock()) {
      priorityStats.add(taskDelay);
      statsManager.trimWindow(priorityStats);
    }
  }
  
  return result;
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
@Override
public final void run() {
  int startRunningCount = currentRunningCount.incrementAndGet();
  
  runTime.addLast(Clock.accurateForwardProgressingMillis());
  try {
    handleRunStart();
  } catch (InterruptedException e) {
    // ignored, just reset status
    Thread.currentThread().interrupt();
  } finally {
    if (runDelayInMillis > 0) {
      TestUtils.sleep(runDelayInMillis);
    }
    
    runCount.incrementAndGet();
    
    try {
      handleRunFinish();
    } finally {
      ranConcurrent = currentRunningCount.decrementAndGet() != 0 || // must be first to ensure decrement is called
                        ranConcurrent || startRunningCount != 1;
    }
  }
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
/**
 * Waits a specified amount of time, or until signalComplete has been called a specified amount 
 * of times, or until a failure occurs.
 * <p>
 * If {@link #waitForTest()} is being called multiple times on the same instance, the 
 * signalComplete count is NOT reset.  So you must either create new instances, or pass in a 
 * larger value for the expected signalComplete count.
 * 
 * @param timeoutInMs Timeout to wait for signalComplete action to occur
 * @param signalCount Amount of signalComplete calls to expect before unblocking
 * @throws InterruptedException Thrown if thread is interrupted while waiting
 * @throws TimeoutException Thrown if timeout occurs without signalComplete being called
 */
public void waitForTest(long timeoutInMs, int signalCount) throws InterruptedException, 
                                                                  TimeoutException {
  long startTime = Clock.accurateForwardProgressingMillis();
  long remainingWaitTime = timeoutInMs;
  synchronized (notifyLock) {
    while (this.signalCount < signalCount && 
           remainingWaitTime > 0 && 
           failure == null) {
      notifyLock.wait(remainingWaitTime);
      
      remainingWaitTime = timeoutInMs - (Clock.accurateForwardProgressingMillis() - startTime);
    }
  }
  
  if (failure != null) {
    throw failure;
  } else if (remainingWaitTime <= 0) {
    throw new TimeoutException();
  }
  // if neither are true we exited normally
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
/**
 * Blocks until the System clock advances at least 1 millisecond.  This will also ensure that 
 * the {@link Clock} class's representation of time has advanced.
 */
public static void blockTillClockAdvances() {
  new TestCondition() {
    private static final short POLL_INTERVAL_IN_MS = 1;
    
    private final long startTime = Clock.accurateTimeMillis();
    private final long alwaysProgressingStartTime = Clock.accurateForwardProgressingMillis();
    
    @Override
    public boolean get() {
      return Clock.accurateTimeMillis() > startTime && 
               Clock.accurateForwardProgressingMillis() > alwaysProgressingStartTime;
    }
    
    @Override
    public void blockTillTrue() {
      blockTillTrue(DEFAULT_TIMEOUT, POLL_INTERVAL_IN_MS);
    }
  }.blockTillTrue();
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
/**
 * Blocks till condition is true, useful for asynchronism operations, waiting for them to 
 * complete in other threads during unit tests.
 * 
 * @param timeoutInMillis time to wait for value to become true
 * @param pollIntervalInMillis time to sleep between checks
 */
public void blockTillTrue(int timeoutInMillis, int pollIntervalInMillis) {
  long startTime = Clock.accurateForwardProgressingMillis();
  long now = startTime;
  boolean lastResult;
  while (! (lastResult = get()) && ! Thread.currentThread().isInterrupted() && 
         (now = Clock.accurateForwardProgressingMillis()) - startTime < timeoutInMillis) {
    if (pollIntervalInMillis > SPIN_THRESHOLD) {
      LockSupport.parkNanos(Clock.NANOS_IN_MILLISECOND * pollIntervalInMillis);
    }
  }
  
  if (! lastResult) {
    throw new ConditionTimeoutException("Still false after " + 
                                          (now - startTime) + "ms, interrupted: " + 
                                          Thread.currentThread().isInterrupted());
  }
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
@Test
public void awaitTerminationTest() throws InterruptedException {
  PrioritySchedulerServiceFactory factory = getPrioritySchedulerFactory();
  try {
    PriorityScheduler scheduler = factory.makePriorityScheduler(1);
    
    TestRunnable tr = new TestRunnable(DELAY_TIME * 2);
    long start = Clock.accurateForwardProgressingMillis();
    scheduler.execute(tr);
    
    tr.blockTillStarted();
    scheduler.shutdown();

    scheduler.awaitTermination();
    long stop = Clock.accurateForwardProgressingMillis();
    
    assertTrue(stop - start >= (DELAY_TIME * 2) - 10);
  } finally {
    factory.shutdown();
  }
}
 

import org.threadly.util.Clock; //导入依赖的package包/类
@Test
public void awaitTerminationTimeoutTest() throws InterruptedException {
  PrioritySchedulerServiceFactory factory = getPrioritySchedulerFactory();
  try {
    PriorityScheduler scheduler = factory.makePriorityScheduler(1);
    
    TestRunnable tr = new TestRunnable(DELAY_TIME * 2);
    long start = Clock.accurateForwardProgressingMillis();
    scheduler.execute(tr);
    
    tr.blockTillStarted();
    scheduler.shutdown();

    assertTrue(scheduler.awaitTermination(DELAY_TIME * 10));
    long stop = Clock.accurateForwardProgressingMillis();
    
    assertTrue(stop - start >= (DELAY_TIME * 2) - 10);
  } finally {
    factory.shutdown();
  }
}