Java 类java.util.concurrent.DelayQueue 实例源码

public void testGet_concurrent() {
  assertTrue(ArbitraryInstances.get(BlockingDeque.class).isEmpty());
  assertTrue(ArbitraryInstances.get(BlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(DelayQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(SynchronousQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(PriorityBlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentMap.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentNavigableMap.class).isEmpty());
  ArbitraryInstances.get(Executor.class).execute(ArbitraryInstances.get(Runnable.class));
  assertNotNull(ArbitraryInstances.get(ThreadFactory.class));
  assertFreshInstanceReturned(
      BlockingQueue.class, BlockingDeque.class, PriorityBlockingQueue.class,
      DelayQueue.class, SynchronousQueue.class,
      ConcurrentMap.class, ConcurrentNavigableMap.class,
      AtomicReference.class, AtomicBoolean.class,
      AtomicInteger.class, AtomicLong.class, AtomicDouble.class);
}

public static void main(String[] args) throws Throwable {

        final DelayQueue<Delayed> q = new DelayQueue<>();
        final long t0 = System.nanoTime();
        for (long i = 0; i < 1000; i++) {
            final long expiry = t0 + i*10L*1000L*1000L;
            q.add(new Delayed() {
                    public long getDelay(TimeUnit unit) {
                        return unit.convert(expiry - System.nanoTime(),
                                            NANOSECONDS);
                    }
                    public int compareTo(Delayed x) {
                        long d = getDelay(NANOSECONDS)
                            - x.getDelay(NANOSECONDS);
                        return d < 0 ? -1 : d > 0 ? 1 : 0; }});
        }

        for (int i = 0; i < 300; i++)
            new Thread() { public void run() {
                try {
                    while (!q.isEmpty())
                        q.poll(10L, TimeUnit.SECONDS);
                } catch (Throwable t) { t.printStackTrace(); }
            }}.start();
    }

private static void realMain(String[] args) throws Throwable {
    Godot[] godots = new Godot[] { new Godot(), new Godot(), new Godot() };
    DelayQueue<Godot> q = new DelayQueue<>(Arrays.asList(godots));
    Iterator<Godot> it = q.iterator();
    q.clear();
    check(it.hasNext());
    equal(it.next(), godots[0]);
    it.remove();
    check(q.isEmpty());

    q.addAll(Arrays.asList(godots));
    it = q.iterator();
    check(it.hasNext());
    it.next();
    equal(it.next(), godots[1]);
    it.remove();
    equal(q.size(), 2);
    check(q.contains(godots[0]));
    check(q.contains(godots[2]));
}

/**
 * retainAll(c) retains only those elements of c and reports true if changed
 */
public void testRetainAll() {
    DelayQueue q = populatedQueue(SIZE);
    DelayQueue p = populatedQueue(SIZE);
    for (int i = 0; i < SIZE; ++i) {
        boolean changed = q.retainAll(p);
        if (i == 0)
            assertFalse(changed);
        else
            assertTrue(changed);

        assertTrue(q.containsAll(p));
        assertEquals(SIZE - i, q.size());
        p.remove();
    }
}

/**
 * timed poll transfers elements across Executor tasks
 */
public void testPollInExecutor() {
    final DelayQueue q = new DelayQueue();
    final CheckedBarrier threadsStarted = new CheckedBarrier(2);
    final ExecutorService executor = Executors.newFixedThreadPool(2);
    try (PoolCleaner cleaner = cleaner(executor)) {
        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                assertNull(q.poll());
                threadsStarted.await();
                assertNotNull(q.poll(LONG_DELAY_MS, MILLISECONDS));
                checkEmpty(q);
            }});

        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                threadsStarted.await();
                q.put(new PDelay(1));
            }});
    }
}

/**
 * Delayed actions do not occur until their delay elapses
 */
public void testDelay() throws InterruptedException {
    DelayQueue<NanoDelay> q = new DelayQueue<>();
    for (int i = 0; i < SIZE; ++i)
        q.add(new NanoDelay(1000000L * (SIZE - i)));

    long last = 0;
    for (int i = 0; i < SIZE; ++i) {
        NanoDelay e = q.take();
        long tt = e.getTriggerTime();
        assertTrue(System.nanoTime() - tt >= 0);
        if (i != 0)
            assertTrue(tt >= last);
        last = tt;
    }
    assertTrue(q.isEmpty());
}

/**
 * drainTo(c) empties queue into another collection c
 */
public void testDrainTo() {
    DelayQueue q = new DelayQueue();
    PDelay[] elems = new PDelay[SIZE];
    for (int i = 0; i < SIZE; ++i) {
        elems[i] = new PDelay(i);
        q.add(elems[i]);
    }
    ArrayList l = new ArrayList();
    q.drainTo(l);
    assertEquals(0, q.size());
    for (int i = 0; i < SIZE; ++i)
        assertEquals(elems[i], l.get(i));
    q.add(elems[0]);
    q.add(elems[1]);
    assertFalse(q.isEmpty());
    assertTrue(q.contains(elems[0]));
    assertTrue(q.contains(elems[1]));
    l.clear();
    q.drainTo(l);
    assertEquals(0, q.size());
    assertEquals(2, l.size());
    for (int i = 0; i < 2; ++i)
        assertEquals(elems[i], l.get(i));
}

public void testGet_concurrent() {
  assertTrue(ArbitraryInstances.get(BlockingDeque.class).isEmpty());
  assertTrue(ArbitraryInstances.get(BlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(DelayQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(SynchronousQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(PriorityBlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentMap.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentNavigableMap.class).isEmpty());
  ArbitraryInstances.get(Executor.class).execute(ArbitraryInstances.get(Runnable.class));
  assertNotNull(ArbitraryInstances.get(ThreadFactory.class));
  assertFreshInstanceReturned(
      BlockingQueue.class, BlockingDeque.class, PriorityBlockingQueue.class,
      DelayQueue.class, SynchronousQueue.class,
      ConcurrentMap.class, ConcurrentNavigableMap.class,
      AtomicReference.class, AtomicBoolean.class,
      AtomicInteger.class, AtomicLong.class, AtomicDouble.class);
}

public static void main(String[] args) throws Throwable {

        final DelayQueue<Delayed> q = new DelayQueue<Delayed>();
        final long t0 = System.nanoTime();
        for (long i = 0; i < 1000; i++) {
            final long expiry = t0 + i*10L*1000L*1000L;
            q.add(new Delayed() {
                    public long getDelay(TimeUnit unit) {
                        return unit.convert(expiry - System.nanoTime(),
                                            NANOSECONDS);
                    }
                    public int compareTo(Delayed x) {
                        long d = getDelay(NANOSECONDS)
                            - x.getDelay(NANOSECONDS);
                        return d < 0 ? -1 : d > 0 ? 1 : 0; }});
        }

        for (int i = 0; i < 300; i++)
            new Thread() { public void run() {
                try {
                    while (!q.isEmpty())
                        q.poll(10L, TimeUnit.SECONDS);
                } catch (Throwable t) { t.printStackTrace(); }
            }}.start();
    }

private static void realMain(String[] args) throws Throwable {
    Godot[] godots = new Godot[] { new Godot(), new Godot(), new Godot() };
    DelayQueue<Godot> q = new DelayQueue<Godot>(Arrays.asList(godots));
    Iterator<Godot> it = q.iterator();
    q.clear();
    check(it.hasNext());
    equal(it.next(), godots[0]);
    it.remove();
    check(q.isEmpty());

    q.addAll(Arrays.asList(godots));
    it = q.iterator();
    check(it.hasNext());
    it.next();
    equal(it.next(), godots[1]);
    it.remove();
    equal(q.size(), 2);
    check(q.contains(godots[0]));
    check(q.contains(godots[2]));
}

/**
 * retainAll(c) retains only those elements of c and reports true if changed
 */
public void testRetainAll() {
    DelayQueue q = populatedQueue(SIZE);
    DelayQueue p = populatedQueue(SIZE);
    for (int i = 0; i < SIZE; ++i) {
        boolean changed = q.retainAll(p);
        if (i == 0)
            assertFalse(changed);
        else
            assertTrue(changed);

        assertTrue(q.containsAll(p));
        assertEquals(SIZE - i, q.size());
        p.remove();
    }
}

/**
 * timed poll transfers elements across Executor tasks
 */
public void testPollInExecutor() {
    final DelayQueue q = new DelayQueue();
    final CheckedBarrier threadsStarted = new CheckedBarrier(2);
    final ExecutorService executor = Executors.newFixedThreadPool(2);
    try (PoolCleaner cleaner = cleaner(executor)) {
        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                assertNull(q.poll());
                threadsStarted.await();
                assertNotNull(q.poll(LONG_DELAY_MS, MILLISECONDS));
                checkEmpty(q);
            }});

        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                threadsStarted.await();
                q.put(new PDelay(1));
            }});
    }
}

/**
 * Delayed actions do not occur until their delay elapses
 */
public void testDelay() throws InterruptedException {
    DelayQueue<NanoDelay> q = new DelayQueue<NanoDelay>();
    for (int i = 0; i < SIZE; ++i)
        q.add(new NanoDelay(1000000L * (SIZE - i)));

    long last = 0;
    for (int i = 0; i < SIZE; ++i) {
        NanoDelay e = q.take();
        long tt = e.getTriggerTime();
        assertTrue(System.nanoTime() - tt >= 0);
        if (i != 0)
            assertTrue(tt >= last);
        last = tt;
    }
    assertTrue(q.isEmpty());
}

/**
 * drainTo(c) empties queue into another collection c
 */
public void testDrainTo() {
    DelayQueue q = new DelayQueue();
    PDelay[] elems = new PDelay[SIZE];
    for (int i = 0; i < SIZE; ++i) {
        elems[i] = new PDelay(i);
        q.add(elems[i]);
    }
    ArrayList l = new ArrayList();
    q.drainTo(l);
    assertEquals(0, q.size());
    for (int i = 0; i < SIZE; ++i)
        assertEquals(elems[i], l.get(i));
    q.add(elems[0]);
    q.add(elems[1]);
    assertFalse(q.isEmpty());
    assertTrue(q.contains(elems[0]));
    assertTrue(q.contains(elems[1]));
    l.clear();
    q.drainTo(l);
    assertEquals(0, q.size());
    assertEquals(2, l.size());
    for (int i = 0; i < 2; ++i)
        assertEquals(elems[i], l.get(i));
}

public HostSelector(Collection<String> baseHosts, String signature)
{
  if (baseHosts.size() < 1)
  {
    throw new IllegalArgumentException("At least one host must be provided.");
  }

  this.fBaseHosts = new TreeSet(baseHosts);
  this.fBlacklist = new DelayQueue();
  this.fIdealHost = null;

  if (signature == null) {
    return;
  }
  int index = Math.abs(signature.hashCode()) % baseHosts.size();

  Iterator it = this.fBaseHosts.iterator();
  while (index-- > 0)
  {
    it.next();
  }
  this.fIdealHost = ((String)it.next());
}

public HostSelector ( Collection<String> baseHosts, String signature )
{
    if ( baseHosts.size () < 1 )
    {
        throw new IllegalArgumentException ( "At least one host must be provided." );
    }

    fBaseHosts = new TreeSet<String> ( baseHosts );
    fBlacklist = new DelayQueue<BlacklistEntry> ();
    fIdealHost = null;

    if ( signature != null )
    {
        // map the signature into an index in the host set
        int index = Math.abs ( signature.hashCode () ) % baseHosts.size();

        // iterate to the selected host
        Iterator<String> it = fBaseHosts.iterator ();
        while ( index-- > 0 )
        {
            it.next ();
        }
        fIdealHost = it.next ();
    }
}

public EntityLruCache ( int maxSize, long maxObjCacheTime, TimeUnit maxObjCacheTimeUnit )
{
    //A load factor > 1 along with a size limit guarantees that the map will not be resized
    super(maxSize, 1.25f, true);

    if (maxSize <= 0)
        throw new IllegalArgumentException("Cache size must be greater than 0");

    this.MAX_ENTRIES = maxSize;     
    this.hits = 0;
    this.misses = 0;

    fMaxAgeMs = TimeUnit.MILLISECONDS.convert ( maxObjCacheTime, maxObjCacheTimeUnit );
    fTimers = new DelayQueue<TimerEntry> ();
    fClock = null;
}

@VisibleForTesting
LocalDataCenterEndPointProvider(CuratorFramework curator,
                                InvalidationServiceEndPointAdapter endPointAdapter,
                                ServiceEndPoint self,
                                MetricRegistry metricRegistry,
                                LifeCycleRegistry lifeCycleRegistry,
                                ExecutorService delayedInvalidationService) {
    _curator = curator;
    _endPointAdapter = endPointAdapter;
    _self = self;
    _metricRegistry = metricRegistry;
    _delayedInvalidationService = delayedInvalidationService;

    _delayedInvalidationQueue = new DelayQueue<>();

    lifeCycleRegistry.manage(this);
}

public NameNodeService() throws IOException {
    URI uri = URI.create(CrailConstants.NAMENODE_ADDRESS);
    String query = uri.getRawQuery();
    StringTokenizer tokenizer = new StringTokenizer(query, "&");
    this.serviceId = Long.parseLong(tokenizer.nextToken().substring(3));
    this.serviceSize = Long.parseLong(tokenizer.nextToken().substring(5));
    this.sequenceId = new AtomicLong(serviceId);
    this.blockStore = new BlockStore();
    this.deleteQueue = new DelayQueue<AbstractNode>();
    this.fileTree = new FileStore(this);
    this.fileTable = new ConcurrentHashMap<Long, AbstractNode>();
    this.gcServer = new GCServer(this, deleteQueue);

    AbstractNode root = fileTree.getRoot();
    fileTable.put(root.getFd(), root);
    Thread gc = new Thread(gcServer);
    gc.start();             
}

public void testGet_concurrent() {
  assertTrue(ArbitraryInstances.get(BlockingDeque.class).isEmpty());
  assertTrue(ArbitraryInstances.get(BlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(DelayQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(SynchronousQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(PriorityBlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentMap.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentNavigableMap.class).isEmpty());
  ArbitraryInstances.get(Executor.class).execute(ArbitraryInstances.get(Runnable.class));
  assertNotNull(ArbitraryInstances.get(ThreadFactory.class));
  assertFreshInstanceReturned(
      BlockingQueue.class, BlockingDeque.class, PriorityBlockingQueue.class,
      DelayQueue.class, SynchronousQueue.class,
      ConcurrentMap.class, ConcurrentNavigableMap.class,
      AtomicReference.class, AtomicBoolean.class,
      AtomicInteger.class, AtomicLong.class, AtomicDouble.class);
}

@SuppressWarnings("unchecked")
protected static <T> Queue<T> createSimilarQueue(Queue<T> orig) {
    if (orig instanceof ArrayBlockingQueue) {
        ArrayBlockingQueue queue = (ArrayBlockingQueue) orig;
        return new ArrayBlockingQueue<T>(queue.size() + queue.remainingCapacity());
    } else if (orig instanceof ArrayDeque) {
        return new ArrayDeque<T>();
    } else if (orig instanceof ConcurrentLinkedQueue) {
        return new ConcurrentLinkedQueue<T>();
    } else if (orig instanceof DelayQueue) {
        return new DelayQueue();
    } else if (orig instanceof LinkedBlockingDeque) {
        return new LinkedBlockingDeque<T>();
    } else if (orig instanceof LinkedBlockingQueue) {
        return new LinkedBlockingQueue<T>();
    } else if (orig instanceof PriorityBlockingQueue) {
        return new PriorityBlockingQueue<T>();
    } else if (orig instanceof PriorityQueue) {
        return new PriorityQueue<T>(11, ((PriorityQueue) orig).comparator());
    } else if (orig instanceof SynchronousQueue) {
        return new SynchronousQueue<T>();
    } else {
        return new LinkedList<T>();
    }
}

/**
 * 
 * typeClone
 * 
 * @param element
 * @return
 */
public static <V, K, T> TypeCloneResult typeCloneQueue(final Queue<V> element) {

    final TypeCloneResult result = new TypeCloneResult();

    final Class clazz = element.getClass();

    if (PriorityQueue.class.isAssignableFrom(clazz)) {
        result.setTypeClone(TypeClone.PriorityQueue);
    } else if (LinkedBlockingQueue.class.isAssignableFrom(clazz)) {
        result.setTypeClone(TypeClone.LinkedBlockingQueue);
    } else if (ArrayBlockingQueue.class.isAssignableFrom(clazz)) {
        result.setTypeClone(TypeClone.ArrayBlockingQueue);
    } else if (PriorityBlockingQueue.class.isAssignableFrom(clazz)) {
        result.setTypeClone(TypeClone.PriorityBlockingQueue);
    } else if (DelayQueue.class.isAssignableFrom(clazz)) {
        result.setTypeClone(TypeClone.DelayQueue);
    } else if (SynchronousQueue.class.isAssignableFrom(clazz)) {
        result.setTypeClone(TypeClone.SynchronousQueue);
    } else {
        result.setTypeClone(TypeClone.NotCloneClass);
    }
    return result;

}

public void testGet_concurrent() {
  assertTrue(ArbitraryInstances.get(BlockingDeque.class).isEmpty());
  assertTrue(ArbitraryInstances.get(BlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(DelayQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(SynchronousQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(PriorityBlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentMap.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentNavigableMap.class).isEmpty());
  ArbitraryInstances.get(Executor.class).execute(ArbitraryInstances.get(Runnable.class));
  assertNotNull(ArbitraryInstances.get(ThreadFactory.class));
  assertFreshInstanceReturned(
      BlockingQueue.class,
      BlockingDeque.class,
      PriorityBlockingQueue.class,
      DelayQueue.class,
      SynchronousQueue.class,
      ConcurrentMap.class,
      ConcurrentNavigableMap.class,
      AtomicReference.class,
      AtomicBoolean.class,
      AtomicInteger.class,
      AtomicLong.class,
      AtomicDouble.class);
}

public void testGet_concurrent() {
  assertTrue(ArbitraryInstances.get(BlockingDeque.class).isEmpty());
  assertTrue(ArbitraryInstances.get(BlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(DelayQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(SynchronousQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(PriorityBlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentMap.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentNavigableMap.class).isEmpty());
  ArbitraryInstances.get(Executor.class).execute(ArbitraryInstances.get(Runnable.class));
  assertNotNull(ArbitraryInstances.get(ThreadFactory.class));
  assertFreshInstanceReturned(
      BlockingQueue.class,
      BlockingDeque.class,
      PriorityBlockingQueue.class,
      DelayQueue.class,
      SynchronousQueue.class,
      ConcurrentMap.class,
      ConcurrentNavigableMap.class,
      AtomicReference.class,
      AtomicBoolean.class,
      AtomicInteger.class,
      AtomicLong.class,
      AtomicDouble.class);
}

/**
 * Execute the task and if reschedule another execution.
 *
 * @param queue
 *     Queue for the pool. This task will be added to the queue to schedule
 *     future executions.
 * @param stats
 *     Handle to stats that should be updated based on the execution of the
 *     task.
 */
@SuppressWarnings("PMD.AvoidCatchingThrowable")
void runAndReschedule(DelayQueue<DelayedTask> queue, Stats stats) {
  thread = Thread.currentThread();
  boolean scheduleAgain = options.schedulingPolicy != Policy.RUN_ONCE;
  try {
    if (!isDone()) {
      task.run();
    }
  } catch (Throwable t) {
    // This catches Throwable because we cannot control the task and thus cannot
    // ensure it is well behaved with respect to exceptions.
    LOGGER.warn("task execution failed", t);
    stats.incrementUncaught(t);
    scheduleAgain = !options.stopOnFailure;
  } finally {
    thread = null;
    if (scheduleAgain && !isDone()) {
      updateNextExecutionTime(stats.skipped());
      queue.put(this);
    } else {
      cancelled = true;
    }
  }
}

public void setRateLimit(int perTenSeconds, int perTenMinutes) {
    if (!usingRateLimiter) {
        perSecondsBucket = new DelayQueue<>();
        for (int i = 0; i < perTenSeconds; i++) {
            perSecondsBucket.put(new Token(TEN_SECONDS, true));
        }
        perMinutesBucket = new DelayQueue<>();
        for (int i = 0; i < perTenMinutes; i++) {
            perMinutesBucket.put(new Token(TEN_MINUTES, true));
        }
        usingRateLimiter = true;
    }
    else {
        throw new IllegalStateException("Can't set rate limit after it has already been set");
    }
}

@Inject
// CHECKSTYLE IGNORE ParameterNumber FOR NEXT 2 LINES
public MqttMessagingSkeleton(@Named(MqttModule.PROPERTY_MQTT_GLOBAL_ADDRESS) MqttAddress ownAddress,
                             @Named(PROPERTY_BACKPRESSURE_REPEATED_MQTT_MESSAGE_IGNORE_PERIOD_MS) int repeatedMqttMessageIgnorePeriodMs,
                             @Named(PROPERTY_BACKPRESSURE_MAX_INCOMING_MQTT_MESSAGES_IN_QUEUE) int maxMqttMessagesInQueue,
                             @Named(PROPERTY_BACKPRESSURE_ENABLED) boolean backpressureEnabled,
                             MessageRouter messageRouter,
                             MqttClientFactory mqttClientFactory,
                             MqttTopicPrefixProvider mqttTopicPrefixProvider,
                             RawMessagingPreprocessor rawMessagingPreprocessor,
                             Set<JoynrMessageProcessor> messageProcessors) {
    this.backpressureEnabled = backpressureEnabled;
    this.ownAddress = ownAddress;
    this.repeatedMqttMessageIgnorePeriodMs = repeatedMqttMessageIgnorePeriodMs;
    this.maxMqttMessagesInQueue = maxMqttMessagesInQueue;
    this.messageRouter = messageRouter;
    this.mqttClientFactory = mqttClientFactory;
    this.mqttTopicPrefixProvider = mqttTopicPrefixProvider;
    this.rawMessagingPreprocessor = rawMessagingPreprocessor;
    this.messageProcessors = messageProcessors;
    this.processingMessages = new HashMap<>();
    this.processedMessagesQueue = new DelayQueue<>();
}

public static void main(String[] args) throws InterruptedException {
    // TODO Auto-generated method stub
    int studentNumber = 20;
    DelayQueue<Student> students = new DelayQueue<Student>();
    Random random = new Random();
    for (int i = 0; i < studentNumber; i++) {
        students.put(new Student("student" + (i + 1), 30 + random.nextInt(120)));
    }
    students.put(new Student("student",120));
    Thread teacherThread = new Thread(new Teacher(students));
    teacherThread.start();
}

/**
 * Create a Task manager with a given number of threads to process the tasks.
 *
 * @param workerCount         maximum number of threads spawned to process the tasks.
 * @param idleTaskDelayMillis delay set for processing a task with IDLE
 *                            {@link org.wso2.broker.core.task.Task.TaskHint}.
 * @param threadFactory       thread factory to be used for processing the tasks.
 */
public TaskExecutorService(int workerCount, long idleTaskDelayMillis, ThreadFactory threadFactory) {

    taskExecutorPool = Executors.newFixedThreadPool(workerCount, threadFactory);
    this.workerCount = workerCount;
    taskProcessorQueue = new ArrayDeque<>(workerCount);
    taskUpdateExecutorService = Executors.newSingleThreadExecutor(threadFactory);
    taskExceptionHandler = new DefaultExceptionHandler();
    taskHolderDelayQueue = new DelayQueue<>();
    taskHolderRegistry = new ConcurrentHashMap<>();
    this.idleTaskDelayMillis = idleTaskDelayMillis;
}

public void init(String nodeIdStr, int memory, int cores,
        int dispatchTime, int heartBeatInterval, ResourceManager rm)
        throws IOException, YarnException {
  super.init(dispatchTime, dispatchTime + 1000000L * heartBeatInterval,
          heartBeatInterval);
  // create resource
  String rackHostName[] = SLSUtils.getRackHostName(nodeIdStr);
  this.node = NodeInfo.newNodeInfo(rackHostName[0], rackHostName[1], 
                BuilderUtils.newResource(memory, cores));
  this.rm = rm;
  // init data structures
  completedContainerList =
          Collections.synchronizedList(new ArrayList<ContainerId>());
  releasedContainerList =
          Collections.synchronizedList(new ArrayList<ContainerId>());
  containerQueue = new DelayQueue<ContainerSimulator>();
  amContainerList =
          Collections.synchronizedList(new ArrayList<ContainerId>());
  runningContainers =
          new ConcurrentHashMap<ContainerId, ContainerSimulator>();
  // register NM with RM
  RegisterNodeManagerRequest req =
          Records.newRecord(RegisterNodeManagerRequest.class);
  req.setNodeId(node.getNodeID());
  req.setResource(node.getTotalCapability());
  req.setHttpPort(80);
  RegisterNodeManagerResponse response = rm.getResourceTrackerService()
          .registerNodeManager(req);
  masterKey = response.getNMTokenMasterKey();
}

void test(String[] args) throws Throwable {
    testDelayQueue(new DelayQueue());
    testDelayQueue(new ScheduledThreadPoolExecutor(1).getQueue());

    testUnbounded(new LinkedBlockingQueue());
    testUnbounded(new LinkedBlockingDeque());
    testUnbounded(new PriorityBlockingQueue());

    testBounded(new LinkedBlockingQueue(CAPACITY));
    testBounded(new LinkedBlockingDeque(CAPACITY));
    testBounded(new ArrayBlockingQueue(CAPACITY));
}

private static void realMain(String[] args) throws Throwable {
    DelayQueue<Godot> q = new DelayQueue<>();
    for (int i = 0; i < 3; i++) {
        equal(q.size(), i);
        equal(q.poll(), null);
        equal(q.size(), i);
        equal(q.poll(100, TimeUnit.MILLISECONDS), null);
        equal(q.size(), i);
        q.add(new Godot());
    }
}

public static Test suite() {
    class Implementation implements CollectionImplementation {
        public Class<?> klazz() { return DelayQueue.class; }
        public Collection emptyCollection() { return new DelayQueue(); }
        public Object makeElement(int i) { return new PDelay(i); }
        public boolean isConcurrent() { return true; }
        public boolean permitsNulls() { return false; }
    }
    return newTestSuite(DelayQueueTest.class,
                        new Generic().testSuite(),
                        CollectionTest.testSuite(new Implementation()));
}

/**
 * Returns a new queue of given size containing consecutive
 * PDelays 0 ... n - 1.
 */
private static DelayQueue<PDelay> populatedQueue(int n) {
    DelayQueue<PDelay> q = new DelayQueue<>();
    assertTrue(q.isEmpty());
    for (int i = n - 1; i >= 0; i -= 2)
        assertTrue(q.offer(new PDelay(i)));
    for (int i = (n & 1); i < n; i += 2)
        assertTrue(q.offer(new PDelay(i)));
    assertFalse(q.isEmpty());
    assertEquals(Integer.MAX_VALUE, q.remainingCapacity());
    assertEquals(n, q.size());
    assertEquals(new PDelay(0), q.peek());
    return q;
}

/**
 * Initializing from null Collection throws NPE
 */
public void testConstructor3() {
    try {
        new DelayQueue(null);
        shouldThrow();
    } catch (NullPointerException success) {}
}

/**
 * Initializing from Collection of null elements throws NPE
 */
public void testConstructor4() {
    try {
        new DelayQueue(Arrays.asList(new PDelay[SIZE]));
        shouldThrow();
    } catch (NullPointerException success) {}
}

/**
 * Initializing from Collection with some null elements throws NPE
 */
public void testConstructor5() {
    PDelay[] a = new PDelay[SIZE];
    for (int i = 0; i < SIZE - 1; ++i)
        a[i] = new PDelay(i);
    try {
        new DelayQueue(Arrays.asList(a));
        shouldThrow();
    } catch (NullPointerException success) {}
}

/**
 * Queue contains all elements of collection used to initialize
 */
public void testConstructor6() {
    PDelay[] ints = new PDelay[SIZE];
    for (int i = 0; i < SIZE; ++i)
        ints[i] = new PDelay(i);
    DelayQueue q = new DelayQueue(Arrays.asList(ints));
    for (int i = 0; i < SIZE; ++i)
        assertEquals(ints[i], q.poll());
}

/**
 * isEmpty is true before add, false after
 */
public void testEmpty() {
    DelayQueue q = new DelayQueue();
    assertTrue(q.isEmpty());
    assertEquals(Integer.MAX_VALUE, q.remainingCapacity());
    q.add(new PDelay(1));
    assertFalse(q.isEmpty());
    q.add(new PDelay(2));
    q.remove();
    q.remove();
    assertTrue(q.isEmpty());
}

/**
 * add succeeds
 */
public void testAdd() {
    DelayQueue q = new DelayQueue();
    for (int i = 0; i < SIZE; ++i) {
        assertEquals(i, q.size());
        assertTrue(q.add(new PDelay(i)));
    }
}