import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
/**
 * Constructs a new parser for a given file.
 * 
 * @param executor Executor that computation can be threaded out to
 * @param hprofFile File that should be parsed
 */
public HprofParser(SubmitterExecutorInterface executor, File hprofFile) {
  ArgumentVerifier.assertNotNull(hprofFile, "hprofFile");
  if (! hprofFile.exists()) {
    throw new IllegalArgumentException("File does not exist: " + hprofFile);
  } else if (! hprofFile.canRead()) {
    throw new IllegalArgumentException("Can not read file: " + hprofFile);
  }
  if (VERBOSE) {  // use single thread in verbose so the out makes sense
    this.executor = SameThreadSubmitterExecutor.instance();
  } else {
    this.executor = executor;
  }
  this.hprofFile = hprofFile;
  parsingFutures = Collections.synchronizedList(new ArrayList<ListenableFuture<?>>());
  classMap = Collections.synchronizedMap(new HashMap<Long, ClassDefinition>());
  instances = Collections.synchronizedMap(new HashMap<Long, Instance>());
  stringMap = new HashMap<>();
  instanceSummary = new HashMap<>();
  arraySummary = new HashMap<>();
  leafInstances = new ArrayList<>();
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
@Override
public ByteBuffer pullBuffer(int size) {
  ArgumentVerifier.assertNotNegative(size, "size");
  if (size == 0) {
    return IOUtils.EMPTY_BYTEBUFFER;
  }
  if (remaining() < size) {
    throw new BufferUnderflowException();
  }
  consumedSize += size;
  final ByteBuffer first = getNextBuffer();
  if(first.remaining() == size) {
    currentBuffer++;
    return first.slice();
  } else if(first.remaining() > size) {
    final ByteBuffer bb = first.duplicate().slice();
    bb.limit(bb.position()+size);
    first.position(first.position()+size);
    return bb;
  } else {
    final byte[] result = new byte[size];
    doGet(result);
    return ByteBuffer.wrap(result);
  }
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
@Override
public void discard(int size) {
  ArgumentVerifier.assertNotNegative(size, "size");
  if (remaining() < size) {
    throw new BufferUnderflowException();
  }
  //We have logic here since we dont need to do any copying and we just drop the bytes
  int toRemoveAmount = size;
  while (toRemoveAmount > 0) {
    final ByteBuffer buf = getNextBuffer();
    final int bufRemaining = buf.remaining();
    if (bufRemaining > toRemoveAmount) {
      buf.position(buf.position() + toRemoveAmount);
      toRemoveAmount = 0;
    } else {
      currentBuffer++;
      toRemoveAmount -= bufRemaining;
    }
  }
  consumedSize += size;
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
@Override
public void discardFromEnd(int size) {
  ArgumentVerifier.assertNotNegative(size, "size");
  if (remaining() < size) {
    throw new BufferUnderflowException();
  }
  //We have logic here since we dont need to do any copying and we just drop the bytes
  int currentIndex = bba.length;
  int toRemoveAmount = size;
  while (toRemoveAmount > 0) {
    final ByteBuffer buf = bba[--currentIndex];
    final int bufRemaining = buf.remaining();
    if (bufRemaining > toRemoveAmount) {
      buf.limit(buf.limit() - toRemoveAmount);
      toRemoveAmount = 0;
    } else {
      bba[currentIndex] = IOUtils.EMPTY_BYTEBUFFER;
      toRemoveAmount -= bufRemaining;
    }
  }
  consumedSize += size;
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
@Override
public ByteBuffer pullBuffer(final int size) {
  ArgumentVerifier.assertNotNegative(size, "size");
  if (size == 0) {
    return IOUtils.EMPTY_BYTEBUFFER;
  }
  if (currentSize < size) {
    throw new BufferUnderflowException();
  }
  consumedSize += size;
  currentSize -= size;
  final ByteBuffer first = availableBuffers.peek();
  if(first.remaining() == size) {
    return removeFirstBuffer().slice();
  } else if(first.remaining() > size) {
    final ByteBuffer bb = first.duplicate().slice();
    bb.limit(bb.position()+size);
    first.position(first.position()+size);
    return bb;
  } else {
    final byte[] result = new byte[size];
    doGet(result);
    return ByteBuffer.wrap(result);
  }
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
@Override
public void discard(final int size) {
  ArgumentVerifier.assertNotNegative(size, "size");
  if (currentSize < size) {
    throw new BufferUnderflowException();
  }
  //We have logic here since we dont need to do any copying and we just drop the bytes
  int toRemoveAmount = size;
  while (toRemoveAmount > 0) {
    final ByteBuffer buf = availableBuffers.peek();
    final int bufRemaining = buf.remaining();
    if (bufRemaining > toRemoveAmount) {
      buf.position(buf.position() + toRemoveAmount);
      toRemoveAmount = 0;
    } else {
      removeFirstBuffer();
      toRemoveAmount -= bufRemaining;
    }
  }
  consumedSize += size;
  currentSize -= size;
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
@Override
public void discardFromEnd(int size) {
  ArgumentVerifier.assertNotNegative(size, "size");
  if (currentSize < size) {
    throw new BufferUnderflowException();
  }
  //We have logic here since we dont need to do any copying and we just drop the bytes
  int toRemoveAmount = size;
  while (toRemoveAmount > 0) {
    final ByteBuffer buf = availableBuffers.peekLast();
    final int bufRemaining = buf.remaining();
    if (bufRemaining > toRemoveAmount) {
      buf.limit(buf.limit() - toRemoveAmount);
      toRemoveAmount = 0;
    } else {
      removeLastBuffer();
      toRemoveAmount -= bufRemaining;
    }
  }
  consumedSize += size;
  currentSize -= size;
}
 

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

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

import org.threadly.util.ArgumentVerifier; //导入依赖的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.ArgumentVerifier; //导入依赖的package包/类
protected WorkerPool(ThreadFactory threadFactory, int poolSize) {
  ArgumentVerifier.assertGreaterThanZero(poolSize, "poolSize");
  if (threadFactory == null) {
    threadFactory = new ConfigurableThreadFactory(PriorityScheduler.class.getSimpleName() + "-", true);
  }
  
  poolSizeChangeLock = new Object();
  idleWorkerDequeLock = new Object();
  idleWorkerCount = new LongAdder();
  idleWorker = new AtomicReference<>(null);
  currentPoolSize = new AtomicInteger(0);
  workerStopNotifyLock = new Object();
  
  this.threadFactory = threadFactory;
  this.maxPoolSize = poolSize;
  this.workerTimedParkRunTime = Long.MAX_VALUE;
  shutdownStarted = new AtomicBoolean(false);
  shutdownFinishing = false;
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
/**
 * Change the set core pool size.  If the value is less than the current max pool size, the max 
 * pool size will also be updated to this value.
 * <p>
 * If this was a reduction from the previous value, this call will examine idle workers to see 
 * if they should be expired.  If this call reduced the max pool size, and the current running 
 * thread count is higher than the new max size, this call will NOT block till the pool is 
 * reduced.  Instead as those workers complete, they will clean up on their own.
 * 
 * @param newPoolSize New core pool size, must be at least one
 */
public void setPoolSize(int newPoolSize) {
  ArgumentVerifier.assertGreaterThanZero(newPoolSize, "newPoolSize");
  
  if (newPoolSize == maxPoolSize) {
    // short cut the lock
    return;
  }
  
  boolean poolSizeIncrease;
  synchronized (poolSizeChangeLock) {
    poolSizeIncrease = newPoolSize > this.maxPoolSize;
    
    this.maxPoolSize = newPoolSize;
  }
  
  handleMaxPoolSizeChange(poolSizeIncrease);
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
/**
 * This will shift an hour from the local time zone to UTC.  This shift will take into account 
 * the current local state of daylight savings time.  The primary usage of this is so that 
 * {@link #getDelayTillHour(int, int)} can be used with a local time zone hour.
 * 
 * @param hour Hour to be shifted in the local time zone in 24 hour format
 * @return Hour shifted to the UTC time zone in 24 hour format
 */
public static int shiftLocalHourToUTC(int hour) {
  ArgumentVerifier.assertLessThan(hour, TimeUnit.DAYS.toHours(1), "hour");
  ArgumentVerifier.assertNotNegative(hour, "hour");
  
  if (cachedHourShift == Integer.MIN_VALUE) {
    Calendar calendar = Calendar.getInstance();
    int shiftInMillis = calendar.get(Calendar.ZONE_OFFSET) + calendar.get(Calendar.DST_OFFSET);
    cachedHourShift = (int)(shiftInMillis / TimeUnit.HOURS.toMillis(1));
  }

  hour -= cachedHourShift;
  if (hour > TimeUnit.DAYS.toHours(1) - 1) {
    hour %= TimeUnit.DAYS.toHours(1);
  } else if (hour < 0) {
    hour += TimeUnit.DAYS.toHours(1);
  }
  
  return hour;
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
/**
 * Schedule a {@link Callable} with a given delay.  This is needed when a result needs to be 
 * consumed from the callable.
 * 
 * @param <T> type of result returned from the future
 * @param threadKey object key where {@code equals()} will be used to determine execution thread
 * @param task callable to be executed
 * @param delayInMs time in milliseconds to wait to execute task
 * @return a future to know when the task has completed and get the result of the callable
 */
public <T> ListenableFuture<T> submitScheduled(Object threadKey, 
                                               Callable<T> task,  long delayInMs) {
  ArgumentVerifier.assertNotNull(threadKey, "threadKey");
  ArgumentVerifier.assertNotNull(task, "task");
  ArgumentVerifier.assertNotNegative(delayInMs, "delayInMs");

  ListenableRunnableFuture<T> rf = new ListenableFutureTask<>(false, task);
  
  if (delayInMs == 0) {
    addTask(threadKey, rf, executor);
  } else {
    scheduler.schedule(new AddTask(threadKey, rf), delayInMs);
  }
  
  return rf;
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
@Override
public void scheduleWithFixedDelay(Runnable task, long initialDelay, long recurringDelay,
                                   TaskPriority priority) {
  ArgumentVerifier.assertNotNull(task, "task");
  ArgumentVerifier.assertNotNegative(initialDelay, "initialDelay");
  ArgumentVerifier.assertNotNegative(recurringDelay, "recurringDelay");
  if (priority == null) {
    priority = defaultPriority;
  }
  
  QueueSet queueSet = queueManager.getQueueSet(priority);
  NoThreadRecurringDelayTaskWrapper taskWrapper = 
      new NoThreadRecurringDelayTaskWrapper(task, queueSet, 
                                            nowInMillis(true) + initialDelay, recurringDelay);
  queueSet.addScheduled(taskWrapper);
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的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.ArgumentVerifier; //导入依赖的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 = queueManager.getQueueSet(priority);
  
  NoThreadRecurringRateTaskWrapper taskWrapper = 
      new NoThreadRecurringRateTaskWrapper(task, queueSet, 
                                           nowInMillis(true) + initialDelay, period);
  queueSet.addScheduled(taskWrapper);
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
protected void setSteps(ExecutionItem[] steps) {
  ArgumentVerifier.assertNotNull(steps, "steps");
  if (this.steps == null) {
    throw new IllegalStateException("Run has completed");
  }
  
  this.steps = steps;
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
protected AbstractScriptStep(CharSequence identifier, ScriptStepType stepType) {
  ArgumentVerifier.assertNotNull(stepType, "stepType");
  
  if (identifier instanceof String) {
    this.identifier = CharsDeduplicator.deDuplicate((String)identifier);
  } else if (identifier == null) {
    this.identifier = "";
  } else {
    this.identifier = identifier;
  }
  
  this.stepType = stepType;
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
/**
 * Constructs a new {@link ExecutableScript}.  If the minimum threads needed don't match the 
 * execution graph provided, it may restrict load, or never complete.  
 * 
 * Execution will not proceed to the next step until the previous step has fully completed.
 * 
 * @param neededThreadQty Minimum number of threads to execute provided steps
 * @param startExecutionItem Execution item which represents the script
 */
public ExecutableScript(int neededThreadQty, ExecutionItem startExecutionItem) {
  if (! startExecutionItem.getChildItems().hasChildren()) {
    throw new IllegalArgumentException("Can not construct script with no steps");
  }
  ArgumentVerifier.assertGreaterThanZero(neededThreadQty, "neededThreadQty");
  
  this.neededThreadQty = neededThreadQty;
  this.startExecutionItem = startExecutionItem;
  scriptAssistant = new ScriptAssistant();
}
 

import org.threadly.util.ArgumentVerifier; //导入依赖的package包/类
/**
 * Initializes the {@link ParameterStore} with the properties to load values used in the 
 * {@link #buildScript()}.  For that reason this should be called right after construction, and 
 * before {@link #buildScript()} is invoked.  This is not done in the constructor because 
 * extending classes should all have an empty (default) constructor.
 * 
 * @param properties Parameters to load values from
 */
protected void initialize(Properties properties) {
  ArgumentVerifier.assertNotNull(properties, "properties");
  if (this.properties != null) {
    throw new IllegalStateException("Alread initialized");
  }
  
  this.properties = properties;
}