Testing an Executor

Question

I have written the following Executor:

/**
 * <p>This executor guarantees that there is never more than one element waiting to be executed. If an element is
 * submitted for execution and another element is still in queue, the newly submitted element replaces the previously
 * queued element in the queue</p>
 *
 * <p>Additionally this executor sleeps a constructor specified number of milliseconds before executing tasks to give queued
 * items a chance to be canceled</p>
 *
 * <p>This class is intended to be used when runnables are created in rapid succession and newer runnables supersede
 * ones that have been created previously. This is e.g. the case if a user interface reacts to user inputs, possibly
 * on every keystroke and the reaction causes slow tasks (e.g. server calls for validation).</p>
 */
public class SingleQueuedElementExecutor implements Executor {
  private final Sleeper sleeper;
  private final ExecutorService delegate;
  private Future<?> queuedSleeper;
  private Future<?> queuedElement;

  public SingleQueuedElementExecutor(final ExecutorService delegate, final long sleepBeforeExecutionMilliseconds) {
    this.delegate = delegate;
    sleeper = new Sleeper(sleepBeforeExecutionMilliseconds);
  }

  public void execute(final Runnable runnable) {
    if (queuedElement != null) {
      queuedElement.cancel(false);
      queuedSleeper.cancel(true);
    }
    queuedSleeper = delegate.submit(sleeper);
    queuedElement = delegate.submit(runnable);
  }

  private static class Sleeper implements Runnable {
    private final long duration;

    private Sleeper(final long duration) {
      this.duration = duration;
    }

    public void run() {
      try {
        Thread.sleep(duration);
      } catch (InterruptedException e) {
        // Simply abort sleeping
      }
    }
  }
}

I believe that the JavaDoc describes accurately what I expect the class to do.

Now the interesting part which I ask a review about is the test for this class I came up with:

public class SingleQueuedElementExecutorTest {
  private SingleQueuedElementExecutor executor;
  private ExecutorService delegateExecutor;
  private static final long SLEEP_BETWEEN_TASKS_MS = 10;
  private static final long NANO_TIME_ACCURACY_MS = 2;

  @BeforeMethod
  public void setUp() throws Exception {
    delegateExecutor = Executors.newSingleThreadExecutor();
    executor = new SingleQueuedElementExecutor(delegateExecutor, SLEEP_BETWEEN_TASKS_MS);
  }

  @Test
  public void execute_sleepsBeforeExecuting() throws Exception {
    // setup
    final InvocationCounter invocationCounter = new InvocationCounter();

    // execution
    final long startTime = System.nanoTime();
    executor.execute(invocationCounter);
    waitForTasksToComplete();
    final long endTime = System.nanoTime();

    // evaluation
    assertThat(endTime - startTime)
        .describedAs("number of ns execution took")
        .isGreaterThanOrEqualTo((SLEEP_BETWEEN_TASKS_MS - NANO_TIME_ACCURACY_MS) * 1000);
  }

  private void waitForTasksToComplete() throws InterruptedException {
    delegateExecutor.shutdown();
    if (!delegateExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
      // There is nothing that takes even near 1 second in this test. So if the executor does not terminate within
      // 30s, than there is surely an error.
      throw new Error("executor did not terminate within 30s.");
    }
  }

  @Test
  public void execute_executesTask() throws Exception {
    // setup
    final InvocationCounter invocationCounter = new InvocationCounter();

    // execution
    executor.execute(invocationCounter);
    waitForTasksToComplete();

    // evaluation
    assertThat(invocationCounter.invocations)
        .isEqualTo(1);
  }

  @Test
  public void execute_abortsPreviouslyAddedTasks() throws Exception {
    // setup
    final InvocationCounter invocationCounterA = new InvocationCounter();
    final InvocationCounter invocationCounterB = new InvocationCounter();

    // execution
    executor.execute(invocationCounterA);
    executor.execute(invocationCounterB);
    waitForTasksToComplete();

    // evaluation
    assertThat(invocationCounterA.invocations)
        .describedAs("invocations of counter A")
        .isEqualTo(0);
    assertThat(invocationCounterB.invocations)
        .describedAs("invocations of counter B")
        .isEqualTo(1);
  }

  private static class InvocationCounter implements Runnable {
    private int invocations = 0;

    public synchronized void run() {
      ++invocations;
    }
  }
}

While the test seems to work fine on my machine I don't like the timing parts of the test. The call to delegateExecutor.awaitTermination(30, TimeUnit.SECONDS) but maybe acceptable. However I strongly dislike my execute_sleepsBeforeExecuting method which relies on System.nanoTime().

What do you think about it? Any suggestions for improvement?

Answer 1

Currently your test is biased to help the unit under test to pass. The delegateExecutor is chosen to be a Executors.newSingleThreadExecutor(), which actually forces the SingleQueuedElementExecutor to finish with the Sleeper task before being able to start the actual task. Replace the delegateExecutor with Executors.newCachedThreadPool() and things get hairier.

Yet the class is helped in another way too : the timing times completion of all tasks, which incidently will include the Sleeper task. If you change the test to time until the start of SingleQueuedElementExecutorTest.InvocationCounter#run, the test will fail consistently.

Also you should note that one millisecond is 1000000 nanoseconds.

Since your unit under test uses Thread.sleep() it's hard to test without doing some stopwatch timing. That being said, it's not so hard to test the behavior you desire. What you basically want is that when the executorService is bombarded in short succession with new tasks that only the most recent one executes. That is actually not so hard to set up, and does not require awkard sleeping in the tests. Another test could check that once a task has begun, a subsequently submitted task will also get executed (if not immediately followed by a slew of tasks itself). This test can simply make use of custom Runnables that force some timing using CountDownLatch instances.

Here's a sample :

public class SingleQueuedElementExecutorTest {

    private SingleQueuedElementExecutor executor;

    private ExecutorService delegateExecutor;
    private static final long SLEEP_BETWEEN_TASKS_MS = 10;
    private static final long NANO_TIME_ACCURACY_MS = 2;

    @Before
    public void setUp() throws Exception {
        delegateExecutor = Executors.newCachedThreadPool();
        executor = new SingleQueuedElementExecutor(delegateExecutor, SLEEP_BETWEEN_TASKS_MS);
    }

    @Test(timeout = 1000)
    public void testOnlyExecuteLastWhenTasksAreSubmittedInQuickSuccession() throws InterruptedException {
        final CountDownLatch lastExecuted = new CountDownLatch(1);
        final AtomicInteger executedCount = new AtomicInteger(0);
        for (int i = 0; i < 1000; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    executedCount.incrementAndGet();
                }
            });
        }
        executor.execute(new Runnable() {
            @Override
            public void run() {
                lastExecuted.countDown();
            }
        });
        lastExecuted.await();
        assertThat(executedCount.get()).isEqualTo(0); // all tasks preceding the last should have been cancelled
    }

    @Test(timeout = 1000)
    public void testSubmissionWhileExecutingPreviousDoesNotCancelPrevious() throws InterruptedException {
        final CountDownLatch startFirst = new CountDownLatch(1);
        final CountDownLatch secondSubmitted = new CountDownLatch(1);
        final CountDownLatch secondDone = new CountDownLatch(1);
        final AtomicBoolean firstExecuted = new AtomicBoolean(false);
        executor.execute(new Runnable() {
            @Override
            public void run() {
                startFirst.countDown();
                try {
                    secondSubmitted.await();
                    firstExecuted.set(true);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });
        startFirst.await();
        executor.execute(new Runnable() {
            @Override
            public void run() {
                secondDone.countDown();
            }
        });
        secondDone.await();
        assertThat(firstExecuted.get()).isTrue();
    }

}

Answer 2

Here is an alternative that you could use that works single threaded and thus completely without timing issues. However it assumes knowledge of internal logic (Name of the class "Sleeper", the way that the executor performs sleeping (by adding a sleeper into the queue) etc.). So neither pretty, but maybe worth a glance:

public class SingleQueuedElementExecutorTest {
  private MockExecutor delegateExecutor;
  private SingleQueuedElementExecutor executor;
  private static final long SLEEP_BETWEEN_TASKS_MS = 10;

  @BeforeMethod
  public void setUp() throws Exception {
    delegateExecutor = new MockExecutor();
    executor = new SingleQueuedElementExecutor(delegateExecutor, SLEEP_BETWEEN_TASKS_MS);
  }

  @Test
  public void execute_sleepsThanExecutes() throws Exception {
    // setup
    final EmptyRunnable emptyRunnable = new EmptyRunnable();

    // execution
    executor.execute(emptyRunnable);

    // evaluation
    assertThat(delegateExecutor.queue)
        .describedAs("queue size")
        .hasSize(2);
    assertThatQueueElementIsSleeper(0, false);
    assertThatQueueElementIsRunnable(1, emptyRunnable, false);
  }

  @Test
  public void execute_abortsPreviouslyAddedTasksAndSleepers() throws Exception {
    // setup
    final EmptyRunnable emptyRunnableA = new EmptyRunnable();
    final EmptyRunnable emptyRunnableB = new EmptyRunnable();

    // execution
    executor.execute(emptyRunnableA);
    executor.execute(emptyRunnableB);

    // evaluation
    assertThat(delegateExecutor.queue)
        .describedAs("queue size")
        .hasSize(4);
    assertThatQueueElementIsSleeper(0, true);
    assertThatQueueElementIsRunnable(1, emptyRunnableA, true);
    assertThatQueueElementIsSleeper(2, false);
    assertThatQueueElementIsRunnable(3, emptyRunnableB, false);
  }

  private void assertThatQueueElementIsSleeper(final int index, final boolean canceled) {
    assertThat(delegateExecutor.queue.get(index).runnable.getClass().getSimpleName())
        .describedAs("queue[" + index + "].runnable.class.name")
        .isEqualTo("Sleeper");
    assertThat(Deencapsulation.getField(delegateExecutor.queue.get(index).runnable, "duration"))
        .describedAs("sleep duration")
        .isEqualTo(SLEEP_BETWEEN_TASKS_MS);
    assertThat(delegateExecutor.queue.get(index).canceled)
        .describedAs("queue[" + index + "].canceled")
        .isEqualTo(canceled);
    if (canceled) {
      assertThat(delegateExecutor.queue.get(index).cancelWithInterrupt)
          .describedAs("queue[" + index + "].cancelWithInterrupt")
          .isTrue();
    }
  }

  private void assertThatQueueElementIsRunnable(final int index, final Runnable runnable, final boolean canceled) {
    assertThat(delegateExecutor.queue.get(index).canceled)
        .describedAs("queue[" + index + "].canceled")
        .isEqualTo(canceled);
    if (canceled) {
      assertThat(delegateExecutor.queue.get(index).cancelWithInterrupt)
          .describedAs("queue[" + index + "].cancelWithInterrupt")
          .isFalse();
    }
    assertThat(delegateExecutor.queue.get(index).runnable)
        .describedAs("queue[" + index + "].runnable")
        .isSameAs(runnable);
  }

  private static class EmptyRunnable implements Runnable {
    public void run() {

    }
  }

  private static class MockExecutor implements ExecutorService {
    private final List<MockFuture<?>> queue = new ArrayList<MockFuture<?>>();

    public void shutdown() {
      throw new UnsupportedOperationException();
    }

    public List<Runnable> shutdownNow() {
      throw new UnsupportedOperationException();
    }

    public boolean isShutdown() {
      throw new UnsupportedOperationException();
    }

    public boolean isTerminated() {
      throw new UnsupportedOperationException();
    }

    public boolean awaitTermination(final long l, final TimeUnit timeUnit) throws InterruptedException {
      throw new UnsupportedOperationException();
    }

    public <T> Future<T> submit(final Callable<T> tCallable) {
      throw new UnsupportedOperationException();
    }

    public <T> Future<T> submit(final Runnable runnable, final T t) {
      throw new UnsupportedOperationException();
    }

    public Future<?> submit(final Runnable runnable) {
      final MockFuture<?> result = new MockFuture<Object>(runnable);
      queue.add(result);
      return result;
    }

    public <T> List<Future<T>> invokeAll(final Collection<? extends Callable<T>> callables) throws InterruptedException {
      throw new UnsupportedOperationException();
    }

    public <T> List<Future<T>> invokeAll(final Collection<? extends Callable<T>> callables, final long l, final TimeUnit timeUnit) throws InterruptedException {
      throw new UnsupportedOperationException();
    }

    public <T> T invokeAny(final Collection<? extends Callable<T>> callables) throws InterruptedException, ExecutionException {
      throw new UnsupportedOperationException();
    }

    public <T> T invokeAny(final Collection<? extends Callable<T>> callables, final long l, final TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException {
      throw new UnsupportedOperationException();
    }

    public void execute(final Runnable runnable) {
      throw new UnsupportedOperationException();
    }
  }

  private static class MockFuture<T> implements Future<T> {
    private boolean canceled;
    private boolean cancelWithInterrupt;
    private final Runnable runnable;

    private MockFuture(final Runnable runnable) {
      this.runnable = runnable;
    }

    public boolean cancel(final boolean b) {
      canceled = true;
      cancelWithInterrupt = b;
      return false;
    }

    public boolean isCancelled() {
      return canceled;
    }

    public boolean isDone() {
      return false;
    }

    public T get() throws InterruptedException, ExecutionException {
      return null;
    }

    public T get(final long l, final TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException {
      return null;
    }
  }
}