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

@BeforeExperiment void setUp() throws Exception {
  executorService = new ThreadPoolExecutor(NUM_THREADS,
      NUM_THREADS,
      Long.MAX_VALUE,
      TimeUnit.SECONDS,
      new ArrayBlockingQueue<Runnable>(1000));
  executorService.prestartAllCoreThreads();
  final AtomicInteger integer = new AtomicInteger();
  // Execute a bunch of tasks to ensure that our threads are allocated and hot
  for (int i = 0; i < NUM_THREADS * 10; i++) {
    @SuppressWarnings("unused") // go/futurereturn-lsc
    Future<?> possiblyIgnoredError =
        executorService.submit(
            new Runnable() {
              @Override
              public void run() {
                integer.getAndIncrement();
              }
            });
  }
}

/**
 * This test tests the CheckedQueue.offer method.
 */
@Test
public void testOffer() {
    ArrayBlockingQueue<String> abq = new ArrayBlockingQueue(1);
    Queue q = Collections.checkedQueue(abq, String.class);

    try {
        q.offer(null);
        fail("should throw NullPointerException.");
    } catch (NullPointerException npe) {
        // Do nothing
    }

    try {
        q.offer(0);
        fail("should throw ClassCastException.");
    } catch (ClassCastException cce) {
        // Do nothing
    }

    assertTrue(q.offer("0"), "queue should have room");

    // no room at the inn!
    assertFalse(q.offer("1"), "queue should be full");
}

public static void main(String[] args) throws Exception {
     final int maxConsumers = (args.length > 0)
         ? Integer.parseInt(args[0])
         : 5;

     pool = Executors.newCachedThreadPool();
     for (int i = 1; i <= maxConsumers; i += (i+1) >>> 1) {
         // Adjust iterations to limit typical single runs to <= 10 ms;
         // Notably, fair queues get fewer iters.
         // Unbounded queues can legitimately OOME if iterations
         // high enough, but we have a sufficiently low limit here.
         run(new ArrayBlockingQueue<Integer>(100), i, 1000);
         run(new LinkedBlockingQueue<Integer>(100), i, 1000);
         run(new LinkedBlockingDeque<Integer>(100), i, 1000);
         run(new LinkedTransferQueue<Integer>(), i, 700);
         run(new PriorityBlockingQueue<Integer>(), i, 1000);
         run(new SynchronousQueue<Integer>(), i, 300);
         run(new SynchronousQueue<Integer>(true), i, 200);
         run(new ArrayBlockingQueue<Integer>(100, true), i, 100);
     }
     pool.shutdown();
     if (! pool.awaitTermination(LONG_DELAY_MS, MILLISECONDS))
         throw new Error();
     pool = null;
}

/**
 * timed poll retrieves elements across Executor threads
 */
public void testPollInExecutor() {
    final ArrayBlockingQueue q = new ArrayBlockingQueue(2);
    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();
                assertSame(one, q.poll(LONG_DELAY_MS, MILLISECONDS));
                checkEmpty(q);
            }});

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

/**
 * Constructor. Creates a new demodulator block reading its samples from the given input queue and
 * returning the buffers to the given output queue. Expects input samples to be at baseband (mixing
 * is done by the scheduler)
 *
 * @param inputQueue    Queue that delivers received baseband signals
 * @param outputQueue   Queue to return used buffers from the inputQueue
 * @param packetSize    Size of the packets in the input queue
 */
public Demodulator (ArrayBlockingQueue<SamplePacket> inputQueue, ArrayBlockingQueue<SamplePacket> outputQueue, int packetSize) {
    // Create internal sample buffers:
    // Note that we create the buffers for the case that there is no downsampling necessary
    // All other cases with input decimation > 1 are also possible because they only need
    // smaller buffers.
    this.quadratureSamples = new SamplePacket(packetSize);

    // Create Audio Sink
    this.audioSink = new AudioSink(packetSize, AUDIO_RATE);

    // Create Decimator block
    // Note that the decimator directly reads from the inputQueue and also returns processed packets to the
    // output queue.
    this.decimator = new Decimator(QUADRATURE_RATE[demodulationMode], packetSize, inputQueue, outputQueue);
}

/**
 * Inserts an element in the specified queue. If the Queue is full, this
 * method removes an element from the head of the Queue and then inserts
 * the element at the tail. It can happen that the an element is removed
 * by another thread in {@link SendWorker#processMessage() processMessage}
 * method before this method attempts to remove an element from the queue.
 * This will cause {@link ArrayBlockingQueue#remove() remove} to throw an
 * exception, which is safe to ignore.
 *
 * Unlike {@link #addToRecvQueue(Message) addToRecvQueue} this method does
 * not need to be synchronized since there is only one thread that inserts
 * an element in the queue and another thread that reads from the queue.
 *
 * @param queue
 *          Reference to the Queue
 * @param buffer
 *          Reference to the buffer to be inserted in the queue
 */
private void addToSendQueue(ArrayBlockingQueue<ByteBuffer> queue,
      ByteBuffer buffer) {
    if (queue.remainingCapacity() == 0) {
        try {
            queue.remove();
        } catch (NoSuchElementException ne) {
            // element could be removed by poll()
            LOG.debug("Trying to remove from an empty " +
                    "Queue. Ignoring exception " + ne);
        }
    }
    try {
        queue.add(buffer);
    } catch (IllegalStateException ie) {
        // This should never happen
        LOG.error("Unable to insert an element in the queue " + ie);
    }
}

/**
 * Constructor. Will create a new Decimator block.
 *
 * @param outputSampleRate      // sample rate to which the incoming samples should be decimated
 * @param packetSize            // packet size of the incoming sample packets
 * @param inputQueue            // queue that delivers incoming sample packets
 * @param inputReturnQueue      // queue to return used input sample packets
 */
public Decimator (int outputSampleRate, int packetSize, ArrayBlockingQueue<SamplePacket> inputQueue,
                  ArrayBlockingQueue<SamplePacket> inputReturnQueue) {
    this.outputSampleRate = outputSampleRate;
    this.packetSize = packetSize;
    this.inputQueue = inputQueue;
    this.inputReturnQueue = inputReturnQueue;

    // Create output queues:
    this.outputQueue = new ArrayBlockingQueue<SamplePacket>(OUTPUT_QUEUE_SIZE);
    this.outputReturnQueue = new ArrayBlockingQueue<SamplePacket>(OUTPUT_QUEUE_SIZE);
    for (int i = 0; i < OUTPUT_QUEUE_SIZE; i++)
        outputReturnQueue.offer(new SamplePacket(packetSize));

    // Create half band filters for downsampling:
    this.inputFilter1 = new HalfBandLowPassFilter(8);
    this.inputFilter2 = new HalfBandLowPassFilter(8);
    this.inputFilter3 = new HalfBandLowPassFilter(8);

    // Create local buffers:
    this.tmpDownsampledSamples = new SamplePacket(packetSize);
}

/**
 * @return that may be used to perform processesing asynchronously.
 */
protected static ExecutorService getExecutorService() {

    if (JDBCSQLXML.executorService == null) {
        int      corePoolSize    = 1;
        int      maximumPoolSize = 10;
        long     keepAliveTime   = 1;
        TimeUnit unit            = TimeUnit.SECONDS;

        JDBCSQLXML.workQueue = new ArrayBlockingQueue<Runnable>(10);
        JDBCSQLXML.executorService = new ThreadPoolExecutor(corePoolSize,
                maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    return executorService;
}

/**
 * submit(callable).get() throws InterruptedException if interrupted
 */
public void testInterruptedSubmit() throws InterruptedException {
    final CountDownLatch submitted    = new CountDownLatch(1);
    final CountDownLatch quittingTime = new CountDownLatch(1);
    final Callable<Void> awaiter = new CheckedCallable<Void>() {
        public Void realCall() throws InterruptedException {
            assertTrue(quittingTime.await(2*LONG_DELAY_MS, MILLISECONDS));
            return null;
        }};
    final ExecutorService p
        = new ThreadPoolExecutor(1,1,60, TimeUnit.SECONDS,
                                 new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(p, quittingTime)) {
        Thread t = newStartedThread(new CheckedInterruptedRunnable() {
            public void realRun() throws Exception {
                Future<Void> future = p.submit(awaiter);
                submitted.countDown();
                future.get();
            }});

        await(submitted);
        t.interrupt();
        awaitTermination(t);
    }
}

/**
 * timed invokeAny(c) returns result of some task
 */
public void testTimedInvokeAny5() throws Exception {
    final ExecutorService e =
        new CustomTPE(2, 2,
                      LONG_DELAY_MS, MILLISECONDS,
                      new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(e)) {
        long startTime = System.nanoTime();
        List<Callable<String>> l = new ArrayList<>();
        l.add(new StringTask());
        l.add(new StringTask());
        String result = e.invokeAny(l, LONG_DELAY_MS, MILLISECONDS);
        assertSame(TEST_STRING, result);
        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
    }
}

/**
 * execute allows the same task to be submitted multiple times, even
 * if rejected
 */
public void testRejectedRecycledTask() throws InterruptedException {
    final int nTasks = 1000;
    final CountDownLatch done = new CountDownLatch(nTasks);
    final Runnable recycledTask = new Runnable() {
        public void run() {
            done.countDown();
        }};
    final ThreadPoolExecutor p =
        new ThreadPoolExecutor(1, 30,
                               60, SECONDS,
                               new ArrayBlockingQueue(30));
    try (PoolCleaner cleaner = cleaner(p)) {
        for (int i = 0; i < nTasks; ++i) {
            for (;;) {
                try {
                    p.execute(recycledTask);
                    break;
                }
                catch (RejectedExecutionException ignore) {}
            }
        }
        // enough time to run all tasks
        await(done, nTasks * SHORT_DELAY_MS);
    }
}

@Override
public void start() throws Exception {
    LOGGER.info("Start called");

    Assert.isTrue(stopped);

    stopped = false;

    queue = new ArrayBlockingQueue<CommandWrapper>(100000);

    // Thread de consommation
    consumationThread = new Thread(this);
    consumationThread.setName("TiPi-Consumer");
    consumationThread.setPriority(Thread.NORM_PRIORITY + 1);
    LOGGER.info("Démarrage du Thread de CommandConsumer ...");
    consumationThread.start();

    if (resumeTipiAtBoot) {
        // Reveille les taches tout de suite
        LOGGER.info("Cold restart TiPi ...");
        addCommand(new ColdRestartCommand());
    } else {
        LOGGER.info("Pas de Cold restart de TiPi");
    }
}

LeaderState(RaftServerImpl server, RaftProperties properties) {
  this.server = server;

  stagingCatchupGap = RaftServerConfigKeys.stagingCatchupGap(properties);
  syncInterval = RaftServerConfigKeys.Rpc.sleepTime(properties);

  final ServerState state = server.getState();
  this.raftLog = state.getLog();
  this.currentTerm = state.getCurrentTerm();
  eventQ = new ArrayBlockingQueue<>(4096);
  processor = new EventProcessor();
  pendingRequests = new PendingRequests(server);

  final RaftConfiguration conf = server.getRaftConf();
  Collection<RaftPeer> others = conf.getOtherPeers(state.getSelfId());
  final Timestamp t = new Timestamp().addTimeMs(-server.getMaxTimeoutMs());
  placeHolderIndex = raftLog.getNextIndex();

  senders = new SenderList(others.stream().map(
      p -> server.newLogAppender(this, p, t, placeHolderIndex, true))
      .toArray(LogAppender[]::new));

  voterLists = divideFollowers(conf);
}

/**
 * timed invokeAny(c) throws NPE if c has null elements
 */
public void testTimedInvokeAny3() throws Exception {
    CountDownLatch latch = new CountDownLatch(1);
    final ExecutorService e =
        new CustomTPE(2, 2,
                      LONG_DELAY_MS, MILLISECONDS,
                      new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(e)) {
        List<Callable<String>> l = new ArrayList<>();
        l.add(latchAwaitingStringTask(latch));
        l.add(null);
        try {
            e.invokeAny(l, randomTimeout(), MILLISECONDS);
            shouldThrow();
        } catch (NullPointerException success) {}
        latch.countDown();
    }
}

public String call(String message) throws IOException, InterruptedException {
    String corrId = UUID.randomUUID().toString();

    AMQP.BasicProperties props = new AMQP.BasicProperties
            .Builder()
            .correlationId(corrId)
            .replyTo(replyQueueName)
            .build();

    channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

    final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);

    channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            if (properties.getCorrelationId().equals(corrId)) {
                response.offer(new String(body, "UTF-8"));
            }
        }
    });

    return response.take();
}

/**
 * @param graphName
 * @param version
 * @throws WaySegmentSerializationException 
 * @throws GraphNotExistsException 
 * @throws InterruptedException 
 */
private void readSegments(String graphName, String version) throws GraphNotExistsException, WaySegmentSerializationException, InterruptedException {
    BlockingQueue<IWaySegment> segmentsQueue = new ArrayBlockingQueue<IWaySegment>(10);
    readDao.readStreetSegments(segmentsQueue, graphName, version);
    log.info("Stored segments:");
    while (!segmentsQueue.isEmpty()) {
        IWaySegment seg = segmentsQueue.poll(10, TimeUnit.MILLISECONDS);
        log.info(seg.toString());
        log.info("XInfo:");
        if (seg.getXInfo() == null || seg.getXInfo().isEmpty()) {
            log.info("empty");
        } else {
            for (IXInfo xinfo : seg.getXInfo()) {
                log.info(xinfo.toString());
            }
        }
    }
}

private void streamBaseXInfos(String graph, String version, InputStream inputStream, boolean isSegmentXInfo)
        throws XInfoNotSupportedException, GraphImportException, GraphStorageException, GraphNotExistsException {
    //First checkk if already another import is running. The singleton serverStatus has to be injected therefore
    if (!serverStatus.registerImport()) {
        throw new GraphImportException("Sorry, system is busy, a graph import is currently executed");
    }
    IBaseSegmentProducer<IBaseSegment> producer = null;
    try {

        BlockingQueue<IBaseSegment> segmentsQueue;

        segmentsQueue = new ArrayBlockingQueue<>(queueSize);

        producer = new BaseSegmentProducerImpl<>(inputFormat, inputStream, segmentsQueue);

        Thread producerThread = new Thread(producer, "basesegment-xinfo-parser-thread");
        producerThread.start();

        List<IBaseSegment> segments = new ArrayList<>();
        while (producerThread.isAlive() || !segmentsQueue.isEmpty()) {
            if (!segmentsQueue.isEmpty()) {
                segments.add(segmentsQueue.poll());
            }
            if (segments.size() >= this.batchSize) {
                this.writeSegments(segments,graph,version,isSegmentXInfo);
                segments.clear();
            }
        }
        this.writeSegments(segments,graph,version,isSegmentXInfo);
    } finally {
        serverStatus.unregisterImport();
        if (producer != null && producer.getException() != null) {
            throw new GraphImportException("Graph could not be imported",producer.getException());
        }
    }
}

/**
 * executor using DiscardOldestPolicy drops oldest task if saturated.
 */
public void testSaturatedExecute_DiscardOldestPolicy() {
    final CountDownLatch done = new CountDownLatch(1);
    LatchAwaiter r1 = awaiter(done);
    LatchAwaiter r2 = awaiter(done);
    LatchAwaiter r3 = awaiter(done);
    final ThreadPoolExecutor p =
        new ThreadPoolExecutor(1, 1,
                               LONG_DELAY_MS, MILLISECONDS,
                               new ArrayBlockingQueue<Runnable>(1),
                               new DiscardOldestPolicy());
    try (PoolCleaner cleaner = cleaner(p, done)) {
        assertEquals(LatchAwaiter.NEW, r1.state);
        assertEquals(LatchAwaiter.NEW, r2.state);
        assertEquals(LatchAwaiter.NEW, r3.state);
        p.execute(r1);
        p.execute(r2);
        assertTrue(p.getQueue().contains(r2));
        p.execute(r3);
        assertFalse(p.getQueue().contains(r2));
        assertTrue(p.getQueue().contains(r3));
    }
    assertEquals(LatchAwaiter.DONE, r1.state);
    assertEquals(LatchAwaiter.NEW, r2.state);
    assertEquals(LatchAwaiter.DONE, r3.state);
}

/**
 * get of element of invokeAll(c) throws exception on failed task
 */
public void testTimedInvokeAll4() throws Exception {
    final ExecutorService e =
        new CustomTPE(2, 2,
                      LONG_DELAY_MS, MILLISECONDS,
                      new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(e)) {
        List<Callable<String>> l = new ArrayList<>();
        l.add(new NPETask());
        List<Future<String>> futures =
            e.invokeAll(l, LONG_DELAY_MS, MILLISECONDS);
        assertEquals(1, futures.size());
        try {
            futures.get(0).get();
            shouldThrow();
        } catch (ExecutionException success) {
            assertTrue(success.getCause() instanceof NullPointerException);
        }
    }
}

/**
 * Tests run() method.
 */
@Test
public void testRun1() throws Exception {
    blockingQueue = new ArrayBlockingQueue(5);
    channel = EasyMock.createMock(Channel.class);
    ospfArea = new OspfAreaImpl();
    lsaWrapper = new LsaWrapperImpl();
    routerLsa = new RouterLsa();
    routerLsa.setLsType(1);
    lsaWrapper.addLsa(OspfLsaType.ROUTER, routerLsa);
    ospfInterface = new OspfInterfaceImpl();
    ospfInterface.setState(OspfInterfaceState.DR);
    lsaWrapper.setOspfInterface(ospfInterface);
    lsaWrapper.setIsSelfOriginated(true);
    lsaHeader = new LsaHeader();
    lsaHeader.setLsType(1);
    lsaWrapper.setLsaHeader(lsaHeader);
    lsaWrapper.setLsaProcessing("refreshLsa");
    lsaWrapper.setLsdbAge(new LsdbAgeImpl(ospfArea));
    blockingQueue.add(lsaWrapper);
    lsaQueueConsumer = new LsaQueueConsumer(blockingQueue, channel, ospfArea);
    lsaQueueConsumer.run();
    assertThat(lsaQueueConsumer, is(notNullValue()));
}

/**
 * get of element of invokeAll(c) throws exception on failed task
 */
public void testInvokeAll4() throws Exception {
    final ExecutorService e =
        new CustomTPE(2, 2,
                      LONG_DELAY_MS, MILLISECONDS,
                      new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(e)) {
        List<Callable<String>> l = new ArrayList<>();
        l.add(new NPETask());
        List<Future<String>> futures = e.invokeAll(l);
        assertEquals(1, futures.size());
        try {
            futures.get(0).get();
            shouldThrow();
        } catch (ExecutionException success) {
            assertTrue(success.getCause() instanceof NullPointerException);
        }
    }
}

Collection<Queue<Boolean>> concurrentQueues() {
    List<Queue<Boolean>> queues = new ArrayList<Queue<Boolean>>();
    queues.add(new ConcurrentLinkedDeque<Boolean>());
    queues.add(new ConcurrentLinkedQueue<Boolean>());
    queues.add(new ArrayBlockingQueue<Boolean>(count, false));
    queues.add(new ArrayBlockingQueue<Boolean>(count, true));
    queues.add(new LinkedBlockingQueue<Boolean>());
    queues.add(new LinkedBlockingDeque<Boolean>());
    queues.add(new LinkedTransferQueue<Boolean>());

    // Following additional implementations are available from:
    // http://gee.cs.oswego.edu/dl/concurrency-interest/index.html
    // queues.add(new SynchronizedLinkedListQueue<Boolean>());

    // Avoid "first fast, second slow" benchmark effect.
    Collections.shuffle(queues);
    return queues;
}

/**
 * isTerminating is not true when running or when terminated
 */
public void testIsTerminating() throws InterruptedException {
    final ThreadPoolExecutor p =
        new ThreadPoolExecutor(1, 1,
                               LONG_DELAY_MS, MILLISECONDS,
                               new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(p)) {
        final CountDownLatch threadStarted = new CountDownLatch(1);
        final CountDownLatch done = new CountDownLatch(1);
        assertFalse(p.isTerminating());
        p.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                assertFalse(p.isTerminating());
                threadStarted.countDown();
                await(done);
            }});
        await(threadStarted);
        assertFalse(p.isTerminating());
        done.countDown();
        try { p.shutdown(); } catch (SecurityException ok) { return; }
        assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS));
        assertTrue(p.isTerminated());
        assertFalse(p.isTerminating());
    }
}

@Override
public void activateOptions() {

    // check whether the configuration is valid first
    try {
        appenderConfig.validate();
    } catch (InvalidAppenderConfigurationException iace) {
        throw new DbAppenederException(iace);
    }

    // set the threshold if there is such
    appenderConfig.setLoggingThreshold(getThreshold());

    // the logging queue
    queue = new ArrayBlockingQueue<LogEventRequest>(getMaxNumberLogEvents());

}

public void garbageCollectionOfUnreachableIterators() {
    boolean fair = rnd.nextBoolean();
    int capacity = rnd.nextInt(1, 10);
    ArrayBlockingQueue q = new ArrayBlockingQueue(capacity, fair);
    randomizePutIndex(q);
    List<Iterator> its = new ArrayList<>();
    for (int i = 0; i < capacity; i++) q.add(i);
    for (int i = 0; i < capacity; i++) its.add(q.iterator());
    assertEquals(attachedIterators(q), its);
    its = null;
    gcAwait(() -> {
        List<Iterator> trackedIterators = trackedIterators(q);
        assertEquals(trackedIterators.size(), capacity);
        for (Iterator x : trackedIterators)
            if (x != null) return false;
        return true;
    });
    Iterator it = q.iterator(); //
    assertEquals(trackedIterators(q), Collections.singletonList(it));
}

private void pushEvent(ApsDvsEvent ev) {
    if(imuLagMs==0){
        heldEvent=ev;
        return;
    }
    ApsDvsEvent ne = new ApsDvsEvent();
    ne.copyFrom(ev);
    if (!eventQueue.offer(ne)) {
        // increase queue size
        ArrayBlockingQueue<ApsDvsEvent> newQueue = new ArrayBlockingQueue<ApsDvsEvent>(eventQueue.size() * 2);
        log.info("increased event queue to " + newQueue.remainingCapacity() + " events");
        newQueue.addAll(eventQueue);
        eventQueue = newQueue;
        eventQueue.offer(ne);
    };
}

@Test
public void testAnyExceptionWrappedInParameterException() {
    class App {
        @Option(names = "-queue", type = String.class, split = ",")
        ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(2);
    }
    try {
        CommandLine.populateCommand(new App(), "-queue a,b,c".split(" "));
        fail("ParameterException expected");
    } catch (ParameterException ex) {
        assertEquals("IllegalStateException: Queue full while processing argument at or before arg[1] 'a,b,c' in [-queue, a,b,c]: java.lang.IllegalStateException: Queue full", ex.getMessage());
    }
}

@Test
public void testDirect() throws InterruptedException {
  final ArrayBlockingQueue<Thread> sync = new ArrayBlockingQueue<>(1);
  TaskExecutors.DIRECT.execute(
      new Runnable() {
        @Override
        public void run() {
          sync.add(Thread.currentThread());
        }
      });
  Thread actual = sync.poll(TIMEOUT_MS, TimeUnit.MILLISECONDS);
  Assert.assertEquals(Thread.currentThread(), actual);
}

public FDTServer(int port) throws Exception {
    hasToRun = new AtomicBoolean(true);

    // We are not very happy to welcome new clients ... so the priority will be lower
    executor = Utils.getStandardExecService("[ Acceptable ServersThreadPool ] ",
            5,
            10,
            new ArrayBlockingQueue<Runnable>(65500),
            Thread.NORM_PRIORITY - 2);
    ssc = ServerSocketChannel.open();
    ssc.configureBlocking(false);

    ss = ssc.socket();

    String listenIP = config.getListenAddress();
    if (listenIP == null) {
        ss.bind(new InetSocketAddress(port));
    }
    else
    {
        ss.bind(new InetSocketAddress(InetAddress.getByName(listenIP), port));
    }

    sel = Selector.open();
    ssc.register(sel, SelectionKey.OP_ACCEPT);

    if (config.isGSIModeEnabled()) {
        FDTGSIServer gsiServer = new FDTGSIServer(config.getGSIPort());
        gsiServer.start();
        logger.log(Level.INFO, "FDT started in GSI mode on port: " + config.getGSIPort());
    }
    // Monitoring & Nice Prnting
    final ScheduledExecutorService monitoringService = Utils.getMonitoringExecService();

    monitoringService.scheduleWithFixedDelay(new FDTServerMonitorTask(), 10, 10, TimeUnit.SECONDS);

    // in SSH mode this is a ACK message for the client to inform it that the server started ok
    // (the server stdout is piped to client through the SSH channel)
    System.out.println("READY");
}

/**
 * containsAll(c) is true when c contains a subset of elements
 */
public void testContainsAll() {
    ArrayBlockingQueue q = populatedQueue(SIZE);
    ArrayBlockingQueue p = new ArrayBlockingQueue(SIZE);
    for (int i = 0; i < SIZE; ++i) {
        assertTrue(q.containsAll(p));
        assertFalse(p.containsAll(q));
        p.add(new Integer(i));
    }
    assertTrue(p.containsAll(q));
}

/**
 * Constructs a new {@link ThrottledExecutor}.
 *
 * @param timeout the minimal time between each execution in milliseconds
 */
public ThrottledExecutor(final int timeout) {
    if (timeout < 0) {
        throw new IllegalArgumentException("The timeout must be a positive integer.");
    }

    this.executor = getExitingExecutorService(new ThreadPoolExecutor(
            THREAD_COUNT, THREAD_COUNT, 0L, TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<>(2, true),
            new DiscardNewestPolicy()));
}

public QueueConnectionPool(String url, String login, String pass, String driver, int initConnections) {
    try {
        Class.forName(driver);
    } catch (Exception e) {
        e.printStackTrace();
    }
    this.url = url;
    this.login = login;
    this.pass = pass;
    pool = new ArrayBlockingQueue<PoolConnection>(16);
    for (int i = 0; i < initConnections; i++) {
        pool.add(getConnection());
    }
}

private void waitForTask() throws Exception {
    if (!DirectByteBufferPool.initInstance(config.getByteBufferSize(), Config.getMaxTakePollIter())) {
        // this is really wrong ... It cannot be already initialized
        throw new FDTProcolException("The buffer pool cannot be already initialized");
    }

    ExecutorService executor = null;
    ServerSocketChannel ssc = null;
    ServerSocket ss = null;
    Selector sel = null;
    try {
        executor = Utils.getStandardExecService("[ Acceptable ServersThreadPool ] ",
                2,
                10,
                new ArrayBlockingQueue<Runnable>(65500),
                Thread.NORM_PRIORITY - 2);
        ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        ss = ssc.socket();
        ss.bind(new InetSocketAddress(config.getPort()));
        sel = Selector.open();
        ssc.register(sel, SelectionKey.OP_ACCEPT);
        System.out.println("READY");
        Utils.waitAndWork(executor, ss, sel, config);
    } finally {
        logger.log(Level.INFO, "[FDT] [ waitForTask ] main loop FINISHED!");
        // close all the stuff
        Utils.closeIgnoringExceptions(ssc);
        Utils.closeIgnoringExceptions(sel);
        Utils.closeIgnoringExceptions(ss);
        if (executor != null) {
            executor.shutdown();
        }
    }
}

WhiteBox() throws ReflectiveOperationException {
    Class<?> qClass = ArrayBlockingQueue.class;
    Class<?> itrClass  = Class.forName(qClass.getName() + "$Itr");
    Class<?> itrsClass = Class.forName(qClass.getName() + "$Itrs");
    Class<?> nodeClass = Class.forName(itrsClass.getName() + "$Node");
    ITRS      = findVarHandle(qClass, "itrs", itrsClass);
    ITEMS     = findVarHandle(qClass, "items", Object[].class);
    TAKEINDEX = findVarHandle(qClass, "takeIndex", int.class);
    PUTINDEX  = findVarHandle(qClass, "putIndex", int.class);
    COUNT     = findVarHandle(qClass, "count", int.class);
    HEAD      = findVarHandle(itrsClass, "head", nodeClass);
    NEXT      = findVarHandle(nodeClass, "next", nodeClass);
    PREVTAKEINDEX = findVarHandle(itrClass, "prevTakeIndex", int.class);
}

/**
 * setMaximumPoolSize(int) throws IllegalArgumentException if
 * given a value less the core pool size
 */
public void testMaximumPoolSizeIllegalArgumentException() {
    final ThreadPoolExecutor p =
        new ThreadPoolExecutor(2, 3,
                               LONG_DELAY_MS, MILLISECONDS,
                               new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(p)) {
        try {
            p.setMaximumPoolSize(1);
            shouldThrow();
        } catch (IllegalArgumentException success) {}
    }
}

/**
 * allowsCoreThreadTimeOut is by default false.
 */
public void testAllowsCoreThreadTimeOut() {
    final ThreadPoolExecutor p =
        new CustomTPE(2, 2,
                      1000, MILLISECONDS,
                      new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(p)) {
        assertFalse(p.allowsCoreThreadTimeOut());
    }
}

public void setItdEventQueueSize(int itdEventQueueSize) {
    getPrefs().putInt("ITDFilter.itdEventQueueSize", itdEventQueueSize);
    getSupport().firePropertyChange("itdEventQueueSize", this.itdEventQueueSize, itdEventQueueSize);
    this.itdEventQueueSize = itdEventQueueSize;
    if (sendITDsToOtherThread) {
        ITDEventQueue = new ArrayBlockingQueue(itdEventQueueSize);
    }
}

public ApmAgent start () {
   logger.info ("Starting ES APM agent");
   initApmInfo();
   apmApiService = createApiClient (RetrofitApmApiService.class);

   errorsQueue = new ArrayBlockingQueue<> (getQueueCapacity (), fairQueue);
   transactionsQueue = new ArrayBlockingQueue<> (getQueueCapacity (), fairQueue);

   // init senders
   errorsSender = startQueue (new ErrorsDataPump (this, errorsQueue));
   transactionsSender = startQueue (new TransactionsDataPump (this, transactionsQueue));

   return this;
}

public void clear_willClearItrs() {
    boolean fair = rnd.nextBoolean();
    int capacity = rnd.nextInt(2, 10);
    ArrayBlockingQueue q = new ArrayBlockingQueue(capacity, fair);
    randomizePutIndex(q);
    List<Iterator> its = new ArrayList<>();
    for (int i = 0; i < capacity; i++)
        assertTrue(q.add(i));
    assertNull(itrs(q));
    for (int i = 0; i < capacity; i++) {
        its.add(q.iterator());
        assertEquals(trackedIterators(q), its);
        q.poll();
        q.add(capacity + i);
    }
    q.clear();
    assertNull(itrs(q));
    int j = 0;
    for (Iterator it : its) {
        assertTrue(isDetached(it));
        if (rnd.nextBoolean()) assertTrue(it.hasNext());
        if (rnd.nextBoolean()) {
            assertEquals(it.next(), j);
            assertIteratorExhausted(it);
        }
        j++;
    }
}

public void afterPropertiesSet() throws Exception {
    executor = new ThreadPoolExecutor(poolSize,
        poolSize,
        0L,
        TimeUnit.MILLISECONDS,
        new ArrayBlockingQueue(poolSize * 4),
        new NamedThreadFactory(WORKER_NAME),
        new ThreadPoolExecutor.CallerRunsPolicy());
}