ExecutorServiceFront extends AbstractExecutorService { @Override public void execute(Runnable task) { workQueue.offer(task); lock.lock(); try { if (!drainerList.isEmpty()) { executor.execute(drainerList.poll()); } } finally { lock.unlock(); } } ExecutorServiceFront(
      BlockingQueue<Runnable> workQueue,
      ExecutorService executor,
      String poolName,
      int maxDrainers,
      long maxTimeSlice,
      TimeUnit maxTimeSliceUnit);  ExecutorServiceFront(
      BlockingQueue<Runnable> workQueue,
      ExecutorService executor,
      int maxDrainers,
      long maxTimeSlice,
      TimeUnit maxTimeSliceUnit);  ExecutorServiceFront(
      BlockingQueue<Runnable> workQueue,
      ExecutorService executor,
      String poolName,
      int maxDrainers);  ExecutorServiceFront(
      BlockingQueue<Runnable> workQueue, ExecutorService executor, int maxDrainers);  ExecutorServiceFront(
      ExecutorService executor, long maxTimeSlice, TimeUnit maxTimeSliceUnit);  ExecutorServiceFront(ExecutorService executor); @Override void shutdown(); @Override synchronized List<Runnable> shutdownNow(); @Override boolean isShutdown(); @Override boolean isTerminated(); @Override boolean awaitTermination(long timeout, TimeUnit unit); @Override void execute(Runnable task);  }
@Test(groups = "fast") public void testMaxDrainer() throws Exception { Assert.assertEquals(mockExecutor.getNumPendingTasks(), 0); executorFront.execute(countTask); Assert.assertEquals(mockExecutor.getNumPendingTasks(), 1); executorFront.execute(countTask); Assert.assertEquals(mockExecutor.getNumPendingTasks(), 1); mockExecutor.drain(); Assert.assertEquals(mockExecutor.getNumPendingTasks(), 0); Assert.assertEquals(count.get(), 2); }
@Test(groups = "fast") public void testConcurrentDrainerAndSubmit() throws Exception { Thread drainingThread = new Thread(mockExecutor::drain); executorFront.execute(latchTask); executorFront.execute(countTask); Assert.assertEquals(mockExecutor.getNumPendingTasks(), 1); drainingThread.start(); Assert.assertEquals(count.get(), 0); executorFront.execute(countTask); latchTask.proceed(); drainingThread.join(); Assert.assertEquals(count.get(), 2); Assert.assertEquals(mockExecutor.getNumPendingTasks(), 0); executorFront.execute(countTask); Assert.assertEquals(mockExecutor.getNumPendingTasks(), 1); }
@Test(groups = "fast") public void testExpiringSingleDrainer() throws Exception { int numTask = 3; for (int i = 0; i < numTask; i++) { executorFront.execute(slowTask); } Assert.assertEquals(mockExecutor.getNumPendingTasks(), 1); for (int i = 0; i < numTask; i++) { mockExecutor.removeHead().run(); Assert.assertEquals(count.get(), i + 1); } count.set(0); for (int i = 0; i < numTask; i++) { executorFront.execute(countTask); } Assert.assertEquals(mockExecutor.getNumPendingTasks(), 1); mockExecutor.removeHead().run(); Assert.assertEquals(count.get(), numTask); DateTimeUtils.setCurrentMillisOffset(0); }
@Test(groups = "fast") public void testExpiringDualDrainer() throws Exception { executorFront2.execute(slowTask); executorFront2.execute(countTask); executorFront2.execute(slowTask); Assert.assertEquals(mockExecutor.getNumPendingTasks(), 2); AnnotatedRunnable drainer2 = mockExecutor.getRunnableList().get(1); mockExecutor.removeHead().run(); Assert.assertEquals(mockExecutor.getNumPendingTasks(), 2); Assert.assertSame(drainer2, mockExecutor.getRunnableList().get(0)); mockExecutor.removeHead().run(); AnnotatedRunnable drainer1 = mockExecutor.getRunnableList().get(0); Assert.assertNotSame(drainer1, drainer2); Assert.assertEquals(count.get(), 3); DateTimeUtils.setCurrentMillisOffset(0); }
@Test(groups = "fast", dataProvider = "getTestExecutors") public void testDyingThreads(Executor executor) throws Exception { int numTasks = NUM_THREADS * 2; CountDownLatch latch = new CountDownLatch(numTasks); for (int i = 0; i < numTasks; i++) { executor.execute( () -> { latch.countDown(); throw new RuntimeException("Expected Failure"); }); } Assert.assertTrue(latch.await(5, TimeUnit.SECONDS)); CountDownLatch latch2 = new CountDownLatch(NUM_THREADS); for (int i = 0; i < NUM_THREADS; i++) { executor.execute(latch2::countDown); } Assert.assertTrue(latch.await(5, TimeUnit.SECONDS)); }
@Test(groups = "fast") public void testTimeExpirationWithEmptyQueue() throws Exception { try { LinkedBlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(); ExecutorServiceFront executorServiceFront = new ExecutorServiceFront(workQueue, mockExecutor, "fuu", 1, 1, TimeUnit.SECONDS); DateTimeUtils.setCurrentMillisFixed(0); executorServiceFront.execute(latchTask); Assert.assertEquals(mockExecutor.getNumPendingTasks(), 1); Thread t = TestUtils.runInThread( () -> { AnnotatedRunnable drainer = mockExecutor.removeHead(); drainer.run(); }, "drainer"); while (!workQueue.isEmpty()) { Thread.sleep(50); } DateTimeUtils.setCurrentMillisFixed(1001); latchTask.proceed(); t.join(); Assert.assertEquals(mockExecutor.getNumPendingTasks(), 0); } finally { DateTimeUtils.setCurrentMillisSystem(); } }
@Test(groups = "fast") public void testRenameThread() throws Exception { LinkedBlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(); ExecutorServiceFront executorServiceFront = new ExecutorServiceFront(workQueue, mockExecutor, "custom-name", 1); executorServiceFront.execute(latchTask); Assert.assertEquals(mockExecutor.getNumPendingTasks(), 1); Thread t = createDrainerThread(); while (!workQueue.isEmpty()) { Thread.sleep(50); } Assert.assertEquals(t.getName(), "custom-name-000"); latchTask.proceed(); t.join(); Assert.assertEquals(t.getName(), "original"); }
@Test(groups = "fast") public void testCreateManyThreads() throws Exception { LinkedBlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(); ExecutorServiceFront executorServiceFront = new ExecutorServiceFront(workQueue, mockExecutor, "custom-name", 2); LatchTask task1 = LatchTask.createPaused(); LatchTask task2 = LatchTask.createPaused(); executorServiceFront.execute(task1); executorServiceFront.execute(task2); ThreadHelper threadHelper = new ThreadHelper(); LoopThread drainerThread1 = createDrainerThread(threadHelper); task1.waitForStart(); LoopThread drainerThread2 = createDrainerThread(threadHelper); task2.waitForStart(); while (!workQueue.isEmpty()) { Thread.sleep(50); } Assert.assertEquals(drainerThread1.getName(), "custom-name-000"); task1.proceed(); task1.await(); Assert.assertEquals(drainerThread2.getName(), "custom-name-001"); task2.proceed(); task2.await(); drainerThread1.join(); drainerThread2.join(); Assert.assertEquals(drainerThread1.getName(), "drainer"); Assert.assertEquals(drainerThread2.getName(), "drainer"); }