### Question:
WorkSync { public AsyncApply applyAsync( W work ) { WorkUnit<Material,W> unit = enqueueWork( work ); Throwable initialThrowable = tryDoWork( unit, 100, false ); return new AsyncApply() { Throwable throwable = initialThrowable; @Override public void await() throws ExecutionException { checkFailure( throwable ); int tryCount = 0; while ( !unit.isDone() ) { tryCount++; checkFailure( throwable = tryDoWork( unit, tryCount, true ) ); } } }; } WorkSync( Material material ); void apply( W work ); AsyncApply applyAsync( W work );  }### Answer:
@Test void mustNotApplyAsyncWorkInParallelUnderStress() throws Exception { int workers = Runtime.getRuntime().availableProcessors() * 5; int iterations = 1_000; int incrementValue = 42; CountDownLatch startLatch = new CountDownLatch( workers ); CountDownLatch endLatch = new CountDownLatch( workers ); AtomicBoolean start = new AtomicBoolean(); Callable<Void> work = () -> { startLatch.countDown(); boolean spin; do { spin = !start.get(); } while ( spin ); ThreadLocalRandom rng = ThreadLocalRandom.current(); List<AsyncApply> asyncs = new ArrayList<>(); for ( int i = 0; i < iterations; i++ ) { asyncs.add( sync.applyAsync( new AddWork( incrementValue ) ) ); if ( rng.nextInt( 10 ) == 0 ) { for ( AsyncApply async : asyncs ) { async.await(); } asyncs.clear(); } } for ( AsyncApply async : asyncs ) { async.await(); } endLatch.countDown(); return null; }; List<Future<Void>> futureList = new ArrayList<>(); for ( int i = 0; i < workers; i++ ) { futureList.add( executor.submit( work ) ); } startLatch.await(); start.set( true ); endLatch.await(); for ( Future<Void> future : futureList ) { future.get(); } assertThat( count.sum(), lessThan( (long) (workers * iterations) ) ); assertThat( sum.sum(), is( (long) (incrementValue * workers * iterations) ) ); }
@Test void mustApplyWorkAsync() throws Exception { AsyncApply a = sync.applyAsync( new AddWork( 10 ) ); a.await(); assertThat( sum.sum(), is( 10L ) ); AsyncApply b = sync.applyAsync( new AddWork( 20 ) ); AsyncApply c = sync.applyAsync( new AddWork( 30 ) ); b.await(); c.await(); assertThat( sum.sum(), is( 60L ) ); }
@Test void mustCombineWorkAsync() throws Exception { makeWorkStuckAtSemaphore( 1 ); AsyncApply a = sync.applyAsync( new AddWork( 1 ) ); AsyncApply b = sync.applyAsync( new AddWork( 1 ) ); AsyncApply c = sync.applyAsync( new AddWork( 1 ) ); semaphore.release( 2 ); a.await(); b.await(); c.await(); assertThat( sum.sum(), is( 4L ) ); assertThat( count.sum(), is( 2L ) ); }
@Test void mustApplyWorkAsyncEvenWhenInterrupted() throws Exception { Thread.currentThread().interrupt(); sync.applyAsync( new AddWork( 10 ) ).await(); assertThat( sum.sum(), is( 10L ) ); assertTrue( Thread.interrupted() ); }