<commit_before>package org.threadly.concurrent;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.threadly.concurrent.future.ListenableFuture;
import org.threadly.util.Clock;

/**
 * <p>An implementation of {@link PriorityScheduledExecutor} which tracks run and usage 
 * statistics.  This is designed for testing and troubleshooting.  It has a little 
 * more overhead from the normal {@link PriorityScheduledExecutor}.</p>
 * 
 * <p>It helps give insight in how long tasks are running, how well the thread pool is 
 * being utilized, as well as execution frequency.</p>
 * 
 * @author jent - <NAME>
 */
public class PrioritySchedulerStatisticTracker extends PriorityScheduledExecutor {
  private static final int MAX_WINDOW_SIZE = 1000;
  
  protected final AtomicInteger totalHighPriorityExecutions;
  protected final AtomicInteger totalLowPriorityExecutions;
  protected final ConcurrentHashMap<Wrapper, Long> runningTasks;
  protected final LinkedList<Long> runTimes;
  protected final LinkedList<Boolean> lowPriorityWorkerAvailable;
  protected final LinkedList<Boolean> highPriorityWorkerAvailable;
  protected final LinkedList<Long> lowPriorityExecutionDelay;
  protected final LinkedList<Long> highPriorityExecutionDelay;
  
  /**
   * Constructs a new thread pool, though no threads will be started 
   * till it accepts it's first request.  This constructs a default 
   * priority of high (which makes sense for most use cases).  
   * It also defaults low priority worker wait as 500ms.  It also  
   * defaults to all newly created threads being daemon threads.
   * 
   * @param corePoolSize pool size that should be maintained
   * @param maxPoolSize maximum allowed thread count
   * @param keepAliveTimeInMs time to wait for a given thread to be idle before killing
   */
  public PrioritySchedulerStatisticTracker(int corePoolSize, int maxPoolSize,
                                           long keepAliveTimeInMs) {
    super(corePoolSize, maxPoolSize, keepAliveTimeInMs);
    
    totalHighPriorityExecutions = new AtomicInteger();
    totalLowPriorityExecutions = new AtomicInteger();
    runningTasks = new ConcurrentHashMap<Wrapper, Long>();
    runTimes = new LinkedList<Long>();
    lowPriorityWorkerAvailable = new LinkedList<Boolean>();
    highPriorityWorkerAvailable = new LinkedList<Boolean>();
    lowPriorityExecutionDelay = new LinkedList<Long>();
    highPriorityExecutionDelay = new LinkedList<Long>();
  }
  
  /**
   * Constructs a new thread pool, though no threads will be started 
   * till it accepts it's first request.  This constructs a default 
   * priority of high (which makes sense for most use cases).  
   * It also defaults low priority worker wait as 500ms.
   * 
   * @param corePoolSize pool size that should be maintained
   * @param maxPoolSize maximum allowed thread count
   * @param keepAliveTimeInMs time to wait for a given thread to be idle before killing
   * @param useDaemonThreads boolean for if newly created threads should be daemon
   */
  public PrioritySchedulerStatisticTracker(int corePoolSize, int maxPoolSize,
                                           long keepAliveTimeInMs, boolean useDaemonThreads) {
    super(corePoolSize, maxPoolSize, keepAliveTimeInMs, useDaemonThreads);
    
    totalHighPriorityExecutions = new AtomicInteger();
    totalLowPriorityExecutions = new AtomicInteger();
    runningTasks = new ConcurrentHashMap<Wrapper, Long>();
    runTimes = new LinkedList<Long>();
    lowPriorityWorkerAvailable = new LinkedList<Boolean>();
    highPriorityWorkerAvailable = new LinkedList<Boolean>();
    lowPriorityExecutionDelay = new LinkedList<Long>();
    highPriorityExecutionDelay = new LinkedList<Long>();
  }
  
  /**
   * Constructs a new thread pool, though no threads will be started 
   * till it accepts it's first request.  This provides the extra
   * parameters to tune what tasks submitted without a priority will be 
   * scheduled as.  As well as the maximum wait for low priority tasks.
   * The longer low priority tasks wait for a worker, the less chance they will
   * have to make a thread.  But it also makes low priority tasks execution time
   * less predictable.
   * 
   * @param corePoolSize pool size that should be maintained
   * @param maxPoolSize maximum allowed thread count
   * @param keepAliveTimeInMs time to wait for a given thread to be idle before killing
   * @param defaultPriority priority to give tasks which do not specify it
   * @param maxWaitForLowPriorityInMs time low priority tasks wait for a worker
   */
  public PrioritySchedulerStatisticTracker(int corePoolSize, int maxPoolSize,
                                           long keepAliveTimeInMs, TaskPriority defaultPriority, 
                                           long maxWaitForLowPriorityInMs) {
    super(corePoolSize, maxPoolSize, keepAliveTimeInMs, 
          defaultPriority, maxWaitForLowPriorityInMs);
    
    totalHighPriorityExecutions = new AtomicInteger();
    totalLowPriorityExecutions = new AtomicInteger();
    runningTasks = new ConcurrentHashMap<Wrapper, Long>();
    runTimes = new LinkedList<Long>();
    lowPriorityWorkerAvailable = new LinkedList<Boolean>();
    highPriorityWorkerAvailable = new LinkedList<Boolean>();
    lowPriorityExecutionDelay = new LinkedList<Long>();
    highPriorityExecutionDelay = new LinkedList<Long>();
  }
  
  /**
   * Constructs a new thread pool, though no threads will be started 
   * till it accepts it's first request.  This provides the extra
   * parameters to tune what tasks submitted without a priority will be 
   * scheduled as.  As well as the maximum wait for low priority tasks.
   * The longer low priority tasks wait for a worker, the less chance they will
   * have to make a thread.  But it also makes low priority tasks execution time
   * less predictable.
   * 
   * @param corePoolSize pool size that should be maintained
   * @param maxPoolSize maximum allowed thread count
   * @param keepAliveTimeInMs time to wait for a given thread to be idle before killing
   * @param defaultPriority priority to give tasks which do not specify it
   * @param maxWaitForLowPriorityInMs time low priority tasks wait for a worker
   * @param useDaemonThreads boolean for if newly created threads should be daemon
   */
  public PrioritySchedulerStatisticTracker(int corePoolSize, int maxPoolSize,
                                           long keepAliveTimeInMs, TaskPriority defaultPriority, 
                                           long maxWaitForLowPriorityInMs, 
                                           final boolean useDaemonThreads) {
    super(corePoolSize, maxPoolSize, keepAliveTimeInMs, 
          defaultPriority, maxWaitForLowPriorityInMs);
    
    totalHighPriorityExecutions = new AtomicInteger();
    totalLowPriorityExecutions = new AtomicInteger();
    runningTasks = new ConcurrentHashMap<Wrapper, Long>();
    runTimes = new LinkedList<Long>();
    lowPriorityWorkerAvailable = new LinkedList<Boolean>();
    highPriorityWorkerAvailable = new LinkedList<Boolean>();
    lowPriorityExecutionDelay = new LinkedList<Long>();
    highPriorityExecutionDelay = new LinkedList<Long>();
  }
  
  /**
   * Constructs a new thread pool, though no threads will be started 
   * till it accepts it's first request.  This provides the extra
   * parameters to tune what tasks submitted without a priority will be 
   * scheduled as.  As well as the maximum wait for low priority tasks.
   * The longer low priority tasks wait for a worker, the less chance they will
   * have to make a thread.  But it also makes low priority tasks execution time
   * less predictable.
   * 
   * @param corePoolSize pool size that should be maintained
   * @param maxPoolSize maximum allowed thread count
   * @param keepAliveTimeInMs time to wait for a given thread to be idle before killing
   * @param defaultPriority priority to give tasks which do not specify it
   * @param maxWaitForLowPriorityInMs time low priority tasks wait for a worker
   * @param threadFactory thread factory for producing new threads within executor
   */
  public PrioritySchedulerStatisticTracker(int corePoolSize, int maxPoolSize,
                                           long keepAliveTimeInMs, TaskPriority defaultPriority, 
                                           long maxWaitForLowPriorityInMs, ThreadFactory threadFactory) {
    super(corePoolSize, maxPoolSize, keepAliveTimeInMs, 
          defaultPriority, maxWaitForLowPriorityInMs, 
          threadFactory);
    
    totalHighPriorityExecutions = new AtomicInteger();
    totalLowPriorityExecutions = new AtomicInteger();
    runningTasks = new ConcurrentHashMap<Wrapper, Long>();
    runTimes = new LinkedList<Long>();
    lowPriorityWorkerAvailable = new LinkedList<Boolean>();
    highPriorityWorkerAvailable = new LinkedList<Boolean>();
    lowPriorityExecutionDelay = new LinkedList<Long>();
    highPriorityExecutionDelay = new LinkedList<Long>();
  }

  @Override
  protected void runHighPriorityTask(TaskWrapper task) throws InterruptedException {
    Worker w = null;
    synchronized (workersLock) {
      if (! isShutdown()) {
        synchronized (highPriorityWorkerAvailable) {
          highPriorityWorkerAvailable.add(! availableWorkers.isEmpty());
          trimList(highPriorityWorkerAvailable);
        }<commit_msg>This resolves #49 ....thus providing the ability to reset the statistics within the tracker.<commit_after>package org.threadly.concurrent;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.threadly.concurrent.future.ListenableFuture;
import org.threadly.util.Clock;

/**
 * <p>An implementation of {@link PriorityScheduledExecutor} which tracks run and usage 
 * statistics.  This is designed for testing and troubleshooting.  It has a little 
 * more overhead from the normal {@link PriorityScheduledExecutor}.</p>
 * 
 * <p>It helps give insight in how long tasks are running, how well the thread pool is 
 * being utilized, as well as execution frequency.</p>
 * 
 * @author jent - <NAME>
 */
public class PrioritySchedulerStatisticTracker extends PriorityScheduledExecutor {
  private static final int MAX_WINDOW_SIZE = 1000;
  
  protected final AtomicInteger totalHighPriorityExecutions;
  protected final AtomicInteger totalLowPriorityExecutions;
  protected final ConcurrentHashMap<Wrapper, Long> runningTasks;
  protected final LinkedList<Long> runTimes;
  protected final LinkedList<Boolean> lowPriorityWorkerAvailable;
  protected final LinkedList<Boolean> highPriorityWorkerAvailable;
  protected final LinkedList<Long> lowPriorityExecutionDelay;
  protected final LinkedList<Long> highPriorityExecutionDelay;
  
  /**
   * Constructs a new thread pool, though no threads will be started 
   * till it accepts it's first request.  This constructs a default 
   * priority of high (which makes sense for most use cases).  
   * It also defaults low priority worker wait as 500ms.  It also  
   * defaults to all newly created threads being daemon threads.
   * 
   * @param corePoolSize pool size that should be maintained
   * @param maxPoolSize maximum allowed thread count
   * @param keepAliveTimeInMs time to wait for a given thread to be idle before killing
   */
  public PrioritySchedulerStatisticTracker(int corePoolSize, int maxPoolSize,
                                           long keepAliveTimeInMs) {
    super(corePoolSize, maxPoolSize, keepAliveTimeInMs);
    
    totalHighPriorityExecutions = new AtomicInteger();
    totalLowPriorityExecutions = new AtomicInteger();
    runningTasks = new ConcurrentHashMap<Wrapper, Long>();
    runTimes = new LinkedList<Long>();
    lowPriorityWorkerAvailable = new LinkedList<Boolean>();
    highPriorityWorkerAvailable = new LinkedList<Boolean>();
    lowPriorityExecutionDelay = new LinkedList<Long>();
    highPriorityExecutionDelay = new LinkedList<Long>();
  }
  
  /**
   * Constructs a new thread pool, though no threads will be started 
   * till it accepts it's first request.  This constructs a default 
   * priority of high (which makes sense for most use cases).  
   * It also defaults low priority worker wait as 500ms.
   * 
   * @param corePoolSize pool size that should be maintained
   * @param maxPoolSize maximum allowed thread count
   * @param keepAliveTimeInMs time to wait for a given thread to be idle before killing
   * @param useDaemonThreads boolean for if newly created threads should be daemon
   */
  public PrioritySchedulerStatisticTracker(int corePoolSize, int maxPoolSize,
                                           long keepAliveTimeInMs, boolean useDaemonThreads) {
    super(corePoolSize, maxPoolSize, keepAliveTimeInMs, useDaemonThreads);
    
    totalHighPriorityExecutions = new AtomicInteger();
    totalLowPriorityExecutions = new AtomicInteger();
    runningTasks = new ConcurrentHashMap<Wrapper, Long>();
    runTimes = new LinkedList<Long>();
    lowPriorityWorkerAvailable = new LinkedList<Boolean>();
    highPriorityWorkerAvailable = new LinkedList<Boolean>();
    lowPriorityExecutionDelay = new LinkedList<Long>();
    highPriorityExecutionDelay = new LinkedList<Long>();
  }
  
  /**
   * Constructs a new thread pool, though no threads will be started 
   * till it accepts it's first request.  This provides the extra
   * parameters to tune what tasks submitted without a priority will be 
   * scheduled as.  As well as the maximum wait for low priority tasks.
   * The longer low priority tasks wait for a worker, the less chance they will
   * have to make a thread.  But it also makes low priority tasks execution time
   * less predictable.
   * 
   * @param corePoolSize pool size that should be maintained
   * @param maxPoolSize maximum allowed thread count
   * @param keepAliveTimeInMs time to wait for a given thread to be idle before killing
   * @param defaultPriority priority to give tasks which do not specify it
   * @param maxWaitForLowPriorityInMs time low priority tasks wait for a worker
   */
  public PrioritySchedulerStatisticTracker(int corePoolSize, int maxPoolSize,
                                           long keepAliveTimeInMs, TaskPriority defaultPriority, 
                                           long maxWaitForLowPriorityInMs) {
    super(corePoolSize, maxPoolSize, keepAliveTimeInMs, 
          defaultPriority, maxWaitForLowPriorityInMs);
    
    totalHighPriorityExecutions = new AtomicInteger();
    totalLowPriorityExecutions = new AtomicInteger();
    runningTasks = new ConcurrentHashMap<Wrapper, Long>();
    runTimes = new LinkedList<Long>();
    lowPriorityWorkerAvailable = new LinkedList<Boolean>();
    highPriorityWorkerAvailable = new LinkedList<Boolean>();
    lowPriorityExecutionDelay = new LinkedList<Long>();
    highPriorityExecutionDelay = new LinkedList<Long>();
  }
  
  /**
   * Constructs a new thread pool, though no threads will be started 
   * till it accepts it's first request.  This provides the extra
   * parameters to tune what tasks submitted without a priority will be 
   * scheduled as.  As well as the maximum wait for low priority tasks.
   * The longer low priority tasks wait for a worker, the less chance they will
   * have to make a thread.  But it also makes low priority tasks execution time
   * less predictable.
   * 
   * @param corePoolSize pool size that should be maintained
   * @param maxPoolSize maximum allowed thread count
   * @param keepAliveTimeInMs time to wait for a given thread to be idle before killing
   * @param defaultPriority priority to give tasks which do not specify it
   * @param maxWaitForLowPriorityInMs time low priority tasks wait for a worker
   * @param useDaemonThreads boolean for if newly created threads should be daemon
   */
  public PrioritySchedulerStatisticTracker(int corePoolSize, int maxPoolSize,
                                           long keepAliveTimeInMs, TaskPriority defaultPriority, 
                                           long maxWaitForLowPriorityInMs, 
                                           final boolean useDaemonThreads) {
    super(corePoolSize, maxPoolSize, keepAliveTimeInMs, 
          defaultPriority, maxWaitForLowPriorityInMs);
    
    totalHighPriorityExecutions = new AtomicInteger();
    totalLowPriorityExecutions = new AtomicInteger();
    runningTasks = new ConcurrentHashMap<Wrapper, Long>();
    runTimes = new LinkedList<Long>();
    lowPriorityWorkerAvailable = new LinkedList<Boolean>();
    highPriorityWorkerAvailable = new LinkedList<Boolean>();
    lowPriorityExecutionDelay = new LinkedList<Long>();
    highPriorityExecutionDelay = new LinkedList<Long>();
  }
  
  /**
   * Constructs a new thread pool, though no threads will be started 
   * till it accepts it's first request.  This provides the extra
   * parameters to tune what tasks submitted without a priority will be 
   * scheduled as.  As well as the maximum wait for low priority tasks.
   * The longer low priority tasks wait for a worker, the less chance they will
   * have to make a thread.  But it also makes low priority tasks execution time
   * less predictable.
   * 
   * @param corePoolSize pool size that should be maintained
   * @param maxPoolSize maximum allowed thread count
   * @param keepAliveTimeInMs time to wait for a given thread to be idle before killing
   * @param defaultPriority priority to give tasks which do not specify it
   * @param maxWaitForLowPriorityInMs time low priority tasks wait for a worker
   * @param threadFactory thread factory for producing new threads within executor
   */
  public PrioritySchedulerStatisticTracker(int corePoolSize, int maxPoolSize,
                                           long keepAliveTimeInMs, TaskPriority defaultPriority, 
                                           long maxWaitForLowPriorityInMs, ThreadFactory threadFactory) {
    super(corePoolSize, maxPoolSize, keepAliveTimeInMs, 
          defaultPriority, maxWaitForLowPriorityInMs, 
          threadFactory);
    
    totalHighPriorityExecutions = new AtomicInteger();
    totalLowPriorityExecutions = new AtomicInteger();
    runningTasks = new ConcurrentHashMap<Wrapper, Long>();
    runTimes = new LinkedList<Long>();
    lowPriorityWorkerAvailable = new LinkedList<Boolean>();
    highPriorityWorkerAvailable = new LinkedList<Boolean>();
    lowPriorityExecutionDelay = new LinkedList<Long>();
    highPriorityExecutionDelay = new LinkedList<Long>();
  }
  
  /**
   * Clears all collected rolling statistics.  These are the statistics 
   * used for averages and are limited by window sizes.
   * 
   * This does NOT reset the total execution counts.
   */
  public void resetCollectedStats() {
    synchronized (runTimes) {
      runTimes.clear();
    }
    synchronized (lowPriorityWorkerAvailable) {
      lowPriorityWorkerAvailable.clear();
    }
    synchronized (lowPriorityExecutionDelay) {
      lowPriorityExecutionDelay.clear();
    }
    synchronized (highPriorityWorkerAvailable) {
      highPriorityWorkerAvailable.clear();
    }
    synchronized (highPriorityExecutionDelay) {
      highPriorityExecutionDelay.clear();
    }
  }

  @Override
  protected void runHighPriorityTask(TaskWrapper task) throws InterruptedException {
    Worker w = null;
    synchronized (workersLock) {
      if (! isShutdown()) {
        synchronized (highPriorityWorkerAvailable) {
          highPriorityWorkerAvailable.add(! availableWorkers.isEmpty());
          trimList(highPriorityWorkerAvailable);
        }
{'fraction_non_alphanumeric': 0.06178716651790142, 'fraction_numerical': 0.0010909451552117426, 'mean_word_length': 3.5544263775971094, 'pattern_counts': {'":': 0, '<': 85, '<?xml version=': 0, '>': 85, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 36, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7062094', 'n_tokens_mistral': 4950, 'n_tokens_neox': 4829, 'n_words': 1849}