@Test public void testCounter() { int maxThreads = 1; BoundedExecutor boundedExecutor = new BoundedExecutor(executorService, maxThreads); int stageTasks = 100_000; int totalTasks = stageTasks * 2; AtomicInteger counter = new AtomicInteger(); CountDownLatch initializeLatch = new CountDownLatch(maxThreads); CountDownLatch startLatch = new CountDownLatch(1); CountDownLatch completeLatch = new CountDownLatch(totalTasks); for (int i = 0; i < stageTasks; i++) { boundedExecutor.execute(() -> { try { initializeLatch.countDown(); Uninterruptibles.awaitUninterruptibly(startLatch); int initialCount = counter.get(); counter.set(initialCount + 1); } finally { completeLatch.countDown(); } }); } assertTrue(Uninterruptibles.awaitUninterruptibly(initializeLatch, 1, TimeUnit.MINUTES)); startLatch.countDown(); for (int i = 0; i < stageTasks; i++) { boundedExecutor.execute(() -> { try { int initialCount = counter.get(); counter.set(initialCount + 1); } finally { completeLatch.countDown(); } }); } assertTrue(Uninterruptibles.awaitUninterruptibly(completeLatch, 1, TimeUnit.MINUTES)); assertEquals(counter.get(), totalTasks); }
@Override public void execute(Runnable task) { checkState(!failed.get(), "BoundedExecutor is in a failed state"); queue.add(task); int size = queueSize.incrementAndGet(); if (size <= maxThreads) { try { coreExecutor.execute(this::drainQueue); } catch (Throwable e) { failed.set(true); log.error("BoundedExecutor state corrupted due to underlying executor failure"); throw e; } } }
BoundedExecutor implements Executor { @Override public void execute(Runnable task) { checkState(!failed.get(), "BoundedExecutor is in a failed state"); queue.add(task); int size = queueSize.incrementAndGet(); if (size <= maxThreads) { try { coreExecutor.execute(this::drainQueue); } catch (Throwable e) { failed.set(true); log.error("BoundedExecutor state corrupted due to underlying executor failure"); throw e; } } } }
BoundedExecutor implements Executor { @Override public void execute(Runnable task) { checkState(!failed.get(), "BoundedExecutor is in a failed state"); queue.add(task); int size = queueSize.incrementAndGet(); if (size <= maxThreads) { try { coreExecutor.execute(this::drainQueue); } catch (Throwable e) { failed.set(true); log.error("BoundedExecutor state corrupted due to underlying executor failure"); throw e; } } } BoundedExecutor(Executor coreExecutor, int maxThreads); }
BoundedExecutor implements Executor { @Override public void execute(Runnable task) { checkState(!failed.get(), "BoundedExecutor is in a failed state"); queue.add(task); int size = queueSize.incrementAndGet(); if (size <= maxThreads) { try { coreExecutor.execute(this::drainQueue); } catch (Throwable e) { failed.set(true); log.error("BoundedExecutor state corrupted due to underlying executor failure"); throw e; } } } BoundedExecutor(Executor coreExecutor, int maxThreads); @Override void execute(Runnable task); }
BoundedExecutor implements Executor { @Override public void execute(Runnable task) { checkState(!failed.get(), "BoundedExecutor is in a failed state"); queue.add(task); int size = queueSize.incrementAndGet(); if (size <= maxThreads) { try { coreExecutor.execute(this::drainQueue); } catch (Throwable e) { failed.set(true); log.error("BoundedExecutor state corrupted due to underlying executor failure"); throw e; } } } BoundedExecutor(Executor coreExecutor, int maxThreads); @Override void execute(Runnable task);  }