common/collect/QueuesTest.java

/*
 * Copyright (C) 2011 The Guava Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.google.common.collect;

import com.google.common.util.concurrent.Uninterruptibles;

import junit.framework.TestCase;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

/**
 * Tests for {@link Queues}.
 *
 * @author Dimitris Andreou
 */

public class QueuesTest extends TestCase {
  /*
   * All the following tests relate to BlockingQueue methods in Queues.
   */

  public static List<BlockingQueue<Object>> blockingQueues() {
    return ImmutableList.<BlockingQueue<Object>>of(
        new LinkedBlockingQueue<Object>(),
        new LinkedBlockingQueue<Object>(10),
        new SynchronousQueue<Object>(),
        new ArrayBlockingQueue<Object>(10),
        new PriorityBlockingQueue<Object>(10, Ordering.arbitrary()));
  }

  private ExecutorService threadPool;

  @Override
  public void setUp() {
    threadPool = Executors.newCachedThreadPool();
  }

  @Override
  public void tearDown() throws InterruptedException {
    // notice that if a Producer is interrupted (a bug), the Producer will go into an infinite
    // loop, which will be noticed here
    threadPool.shutdown();
    assertTrue("Some worker didn't finish in time",
        threadPool.awaitTermination(1, TimeUnit.SECONDS));
  }

  private static <T> int drain(BlockingQueue<T> q, Collection<? super T> buffer, int maxElements,
      long timeout, TimeUnit unit, boolean interruptibly) throws InterruptedException {
    return interruptibly
        ? Queues.drain(q, buffer, maxElements, timeout, unit)
        : Queues.drainUninterruptibly(q, buffer, maxElements, timeout, unit);
  }

  public void testMultipleProducers() throws Exception {
    for (BlockingQueue<Object> q : blockingQueues()) {
      testMultipleProducers(q);
    }
  }

  private void testMultipleProducers(BlockingQueue<Object> q)
      throws InterruptedException {
    for (boolean interruptibly : new boolean[] { true, false }) {
      threadPool.submit(new Producer(q, 20));
      threadPool.submit(new Producer(q, 20));
      threadPool.submit(new Producer(q, 20));
      threadPool.submit(new Producer(q, 20));
      threadPool.submit(new Producer(q, 20));

      List<Object> buf = Lists.newArrayList();
      int elements = drain(q, buf, 100, Long.MAX_VALUE, TimeUnit.NANOSECONDS, interruptibly);
      assertEquals(100, elements);
      assertEquals(100, buf.size());
      assertDrained(q);
    }
  }

  public void testDrainTimesOut() throws Exception {
    for (BlockingQueue<Object> q : blockingQueues()) {
      testDrainTimesOut(q);
    }
  }

  private void testDrainTimesOut(BlockingQueue<Object> q) throws Exception {
    for (boolean interruptibly : new boolean[] { true, false }) {
      assertEquals(0, Queues.drain(q, ImmutableList.of(), 1, 10, TimeUnit.MILLISECONDS));

      // producing one, will ask for two
      Future<?> submitter = threadPool.submit(new Producer(q, 1));

      // make sure we time out
      long startTime = System.nanoTime();

      int drained = drain(q, Lists.newArrayList(), 2, 10, TimeUnit.MILLISECONDS, interruptibly);
      assertTrue(drained <= 1);

      assertTrue((System.nanoTime() - startTime) >= TimeUnit.MILLISECONDS.toNanos(10));

      // If even the first one wasn't there, clean up so that the next test doesn't see an element.
      submitter.get();
      if (drained == 0) {
        assertNotNull(q.poll());
      }
    }
  }

  public void testZeroElements() throws Exception {
    for (BlockingQueue<Object> q : blockingQueues()) {
      testZeroElements(q);
    }
  }

  private void testZeroElements(BlockingQueue<Object> q) throws InterruptedException {
    for (boolean interruptibly : new boolean[] { true, false }) {
      // asking to drain zero elements
      assertEquals(0, drain(q, ImmutableList.of(), 0, 10, TimeUnit.MILLISECONDS, interruptibly));
    }
  }

  public void testEmpty() throws Exception {
    for (BlockingQueue<Object> q : blockingQueues()) {
      testEmpty(q);
    }
  }

  private void testEmpty(BlockingQueue<Object> q) {
    assertDrained(q);
  }

  public void testNegativeMaxElements() throws Exception {
    for (BlockingQueue<Object> q : blockingQueues()) {
      testNegativeMaxElements(q);
    }
  }

  private void testNegativeMaxElements(BlockingQueue<Object> q) throws InterruptedException {
    threadPool.submit(new Producer(q, 1));

    List<Object> buf = Lists.newArrayList();
    int elements = Queues.drain(q, buf, -1, Long.MAX_VALUE, TimeUnit.NANOSECONDS);
    assertEquals(elements, 0);
    assertTrue(buf.isEmpty());

    // Clean up produced element to free the producer thread, otherwise it will complain
    // when we shutdown the threadpool.
    Queues.drain(q, buf, 1, Long.MAX_VALUE, TimeUnit.NANOSECONDS);
  }

  public void testDrain_throws() throws Exception {
    for (BlockingQueue<Object> q : blockingQueues()) {
      testDrain_throws(q);
    }
  }

  private void testDrain_throws(BlockingQueue<Object> q) {
    threadPool.submit(new Interrupter(Thread.currentThread()));
    try {
      Queues.drain(q, ImmutableList.of(), 100, Long.MAX_VALUE, TimeUnit.NANOSECONDS);
      fail();
    } catch (InterruptedException expected) {
    }
  }

  public void testDrainUninterruptibly_doesNotThrow() throws Exception {
    for (BlockingQueue<Object> q : blockingQueues()) {
      testDrainUninterruptibly_doesNotThrow(q);
    }
  }

  private void testDrainUninterruptibly_doesNotThrow(final BlockingQueue<Object> q) {
    final Thread mainThread = Thread.currentThread();
    threadPool.submit(new Runnable() {
      public void run() {
        new Producer(q, 50).run();
        new Interrupter(mainThread).run();
        new Producer(q, 50).run();
      }
    });
    List<Object> buf = Lists.newArrayList();
    int elements =
        Queues.drainUninterruptibly(q, buf, 100, Long.MAX_VALUE, TimeUnit.NANOSECONDS);
    // so when this drains all elements, we know the thread has also been interrupted in between
    assertTrue(Thread.interrupted());
    assertEquals(100, elements);
    assertEquals(100, buf.size());
  }

  public void testNewLinkedBlockingDequeCapacity() {
    try {
      Queues.newLinkedBlockingDeque(0);
      fail("Should have thrown IllegalArgumentException");
    } catch (IllegalArgumentException expected) {
      // any capacity less than 1 should throw IllegalArgumentException
    }
    assertEquals(1, Queues.newLinkedBlockingDeque(1).remainingCapacity());
    assertEquals(11, Queues.newLinkedBlockingDeque(11).remainingCapacity());
  }

  public void testNewLinkedBlockingQueueCapacity() {
    try {
      Queues.newLinkedBlockingQueue(0);
      fail("Should have thrown IllegalArgumentException");
    } catch (IllegalArgumentException expected) {
      // any capacity less than 1 should throw IllegalArgumentException
    }
    assertEquals(1, Queues.newLinkedBlockingQueue(1).remainingCapacity());
    assertEquals(11, Queues.newLinkedBlockingQueue(11).remainingCapacity());
  }

  /**
   * Checks that #drain() invocations behave correctly for a drained (empty) queue.
   */
  private void assertDrained(BlockingQueue<Object> q) {
    assertNull(q.peek());
    assertInterruptibleDrained(q);
    assertUninterruptibleDrained(q);
  }

  private void assertInterruptibleDrained(BlockingQueue<Object> q) {
    // nothing to drain, thus this should wait doing nothing
    try {
      assertEquals(0, Queues.drain(q, ImmutableList.of(), 0, 10, TimeUnit.MILLISECONDS));
    } catch (InterruptedException e) {
      throw new AssertionError();
    }

    // but does the wait actually occurs?
    threadPool.submit(new Interrupter(Thread.currentThread()));
    try {
      // if waiting works, this should get stuck
      Queues.drain(q, Lists.newArrayList(), 1, Long.MAX_VALUE, TimeUnit.NANOSECONDS);
      fail();
    } catch (InterruptedException expected) {
      // we indeed waited; a slow thread had enough time to interrupt us
    }
  }

  // same as above; uninterruptible version
  private void assertUninterruptibleDrained(BlockingQueue<Object> q) {
    assertEquals(0,
        Queues.drainUninterruptibly(q, ImmutableList.of(), 0, 10, TimeUnit.MILLISECONDS));

    // but does the wait actually occurs?
    threadPool.submit(new Interrupter(Thread.currentThread()));

    long startTime = System.nanoTime();
    Queues.drainUninterruptibly(
        q, Lists.newArrayList(), 1, 10, TimeUnit.MILLISECONDS);
    assertTrue((System.nanoTime() - startTime) >= TimeUnit.MILLISECONDS.toNanos(10));
    // wait for interrupted status and clear it
    while (!Thread.interrupted()) { Thread.yield(); }
  }

  private static class Producer implements Runnable {
    final BlockingQueue<Object> q;
    final int elements;

    Producer(BlockingQueue<Object> q, int elements) {
      this.q = q;
      this.elements = elements;
    }

    @Override public void run() {
      try {
        for (int i = 0; i < elements; i++) {
          q.put(new Object());
        }
      } catch (InterruptedException e) {
        // TODO(user): replace this when there is a better way to spawn threads in tests and
        // have threads propagate their errors back to the test thread.
        e.printStackTrace();
        // never returns, so that #tearDown() notices that one worker isn't done
        Uninterruptibles.sleepUninterruptibly(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
      }
    }
  }

  private static class Interrupter implements Runnable {
    final Thread threadToInterrupt;

    Interrupter(Thread threadToInterrupt) {
      this.threadToInterrupt = threadToInterrupt;
    }

    @Override public void run() {
      try {
        Thread.sleep(100);
      } catch (InterruptedException e) {
        throw new AssertionError();
      } finally {
        threadToInterrupt.interrupt();
      }
    }
  }
}