AsyncSemaphore { public ListenableFuture<?> submit(T task) { QueuedTask<T> queuedTask = new QueuedTask<>(task); queuedTasks.add(queuedTask); acquirePermit(); return queuedTask.getCompletionFuture(); } AsyncSemaphore(int maxPermits, Executor submitExecutor, Function<T, ListenableFuture<?>> submitter); ListenableFuture<?> submit(T task);  }
@Test public void testInlineExecution() throws Exception { AsyncSemaphore<Runnable> asyncSemaphore = new AsyncSemaphore<>(1, executor, task -> newDirectExecutorService().submit(task)); final AtomicInteger count = new AtomicInteger(); List<ListenableFuture<?>> futures = new ArrayList<>(); for (int i = 0; i < 1000; i++) { futures.add(asyncSemaphore.submit(count::incrementAndGet)); } Futures.allAsList(futures).get(1, TimeUnit.MINUTES); Assert.assertEquals(count.get(), 1000); }
@Test public void testSingleThreadBoundedConcurrency() throws Exception { AsyncSemaphore<Runnable> asyncSemaphore = new AsyncSemaphore<>(1, executor, executor::submit); final AtomicInteger count = new AtomicInteger(); final AtomicInteger concurrency = new AtomicInteger(); List<ListenableFuture<?>> futures = new ArrayList<>(); for (int i = 0; i < 1000; i++) { futures.add(asyncSemaphore.submit(() -> { count.incrementAndGet(); int currentConcurrency = concurrency.incrementAndGet(); assertLessThanOrEqual(currentConcurrency, 1); Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS); concurrency.decrementAndGet(); })); } Futures.allAsList(futures).get(1, TimeUnit.MINUTES); Assert.assertEquals(count.get(), 1000); }
@Test public void testMultiThreadBoundedConcurrency() throws Exception { AsyncSemaphore<Runnable> asyncSemaphore = new AsyncSemaphore<>(2, executor, executor::submit); final AtomicInteger count = new AtomicInteger(); final AtomicInteger concurrency = new AtomicInteger(); List<ListenableFuture<?>> futures = new ArrayList<>(); for (int i = 0; i < 1000; i++) { futures.add(asyncSemaphore.submit(() -> { count.incrementAndGet(); int currentConcurrency = concurrency.incrementAndGet(); assertLessThanOrEqual(currentConcurrency, 2); Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS); concurrency.decrementAndGet(); })); } Futures.allAsList(futures).get(1, TimeUnit.MINUTES); Assert.assertEquals(count.get(), 1000); }
@Test public void testMultiSubmitters() { final AsyncSemaphore<Runnable> asyncSemaphore = new AsyncSemaphore<>(2, executor, executor::submit); final AtomicInteger count = new AtomicInteger(); final AtomicInteger concurrency = new AtomicInteger(); final CountDownLatch startLatch = new CountDownLatch(1); final CountDownLatch completionLatch = new CountDownLatch(100); for (int i = 0; i < 100; i++) { executor.execute(() -> { Uninterruptibles.awaitUninterruptibly(startLatch, 1, TimeUnit.MINUTES); asyncSemaphore.submit(() -> { count.incrementAndGet(); int currentConcurrency = concurrency.incrementAndGet(); assertLessThanOrEqual(currentConcurrency, 2); Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS); concurrency.decrementAndGet(); completionLatch.countDown(); }); }); } startLatch.countDown(); Uninterruptibles.awaitUninterruptibly(completionLatch, 1, TimeUnit.MINUTES); Assert.assertEquals(count.get(), 100); }
@Test public void testFailedTasks() throws Exception { AsyncSemaphore<Runnable> asyncSemaphore = new AsyncSemaphore<>(2, executor, executor::submit); final AtomicInteger successCount = new AtomicInteger(); final AtomicInteger failureCount = new AtomicInteger(); final AtomicInteger concurrency = new AtomicInteger(); final CountDownLatch completionLatch = new CountDownLatch(1000); List<ListenableFuture<?>> futures = new ArrayList<>(); for (int i = 0; i < 1000; i++) { ListenableFuture<?> future = asyncSemaphore.submit(() -> assertFailedConcurrency(concurrency)); addCallback(future, completionCallback(successCount, failureCount, completionLatch), directExecutor()); futures.add(future); } completionLatch.await(1, TimeUnit.MINUTES); for (ListenableFuture<?> future : futures) { try { future.get(); Assert.fail(); } catch (Exception ignored) { } } Assert.assertEquals(successCount.get(), 0); Assert.assertEquals(failureCount.get(), 1000); }
@Test public void testFailedTaskSubmission() throws Exception { final AtomicInteger successCount = new AtomicInteger(); final AtomicInteger failureCount = new AtomicInteger(); final AtomicInteger concurrency = new AtomicInteger(); final CountDownLatch completionLatch = new CountDownLatch(1000); AsyncSemaphore<Runnable> asyncSemaphore = new AsyncSemaphore<>(2, executor, task -> { throw assertFailedConcurrency(concurrency); }); List<ListenableFuture<?>> futures = new ArrayList<>(); for (int i = 0; i < 1000; i++) { ListenableFuture<?> future = asyncSemaphore.submit(Assert::fail); addCallback(future, completionCallback(successCount, failureCount, completionLatch), directExecutor()); futures.add(future); } completionLatch.await(1, TimeUnit.MINUTES); for (ListenableFuture<?> future : futures) { try { future.get(); Assert.fail(); } catch (Exception ignored) { } } Assert.assertEquals(successCount.get(), 0); Assert.assertEquals(failureCount.get(), 1000); }
@Test public void testFailedTaskWithMultipleSubmitters() { final AtomicInteger successCount = new AtomicInteger(); final AtomicInteger failureCount = new AtomicInteger(); final AtomicInteger concurrency = new AtomicInteger(); final CountDownLatch startLatch = new CountDownLatch(1); final CountDownLatch completionLatch = new CountDownLatch(100); AsyncSemaphore<Runnable> asyncSemaphore = new AsyncSemaphore<>(2, executor, task -> { throw assertFailedConcurrency(concurrency); }); final Queue<ListenableFuture<?>> futures = new ConcurrentLinkedQueue<>(); for (int i = 0; i < 100; i++) { executor.execute(() -> { Uninterruptibles.awaitUninterruptibly(startLatch, 1, TimeUnit.MINUTES); ListenableFuture<?> future = asyncSemaphore.submit(Assert::fail); futures.add(future); addCallback(future, completionCallback(successCount, failureCount, completionLatch), directExecutor()); }); } startLatch.countDown(); Uninterruptibles.awaitUninterruptibly(completionLatch, 1, TimeUnit.MINUTES); for (ListenableFuture<?> future : futures) { try { future.get(); Assert.fail(); } catch (Exception ignored) { } } Assert.assertEquals(successCount.get(), 0); Assert.assertEquals(failureCount.get(), 100); }
@Test public void testNoStackOverflow() throws Exception { AsyncSemaphore<Object> asyncSemaphore = new AsyncSemaphore<>(1, executor, object -> Futures.immediateFuture(null)); List<ListenableFuture<?>> futures = new ArrayList<>(); for (int i = 0; i < 1000; i++) { futures.add(asyncSemaphore.submit(new Object())); } Futures.allAsList(futures).get(1, TimeUnit.MINUTES); }