### Question:
WorkSync { public void apply( W work ) throws ExecutionException { WorkUnit<Material,W> unit = enqueueWork( work ); int tryCount = 0; do { tryCount++; checkFailure( tryDoWork( unit, tryCount, true ) ); } while ( !unit.isDone() ); } WorkSync( Material material ); void apply( W work ); AsyncApply applyAsync( W work );  }### Answer:
@Test void mustApplyWork() throws Exception { sync.apply( new AddWork( 10 ) ); assertThat( sum.sum(), is( 10L ) ); sync.apply( new AddWork( 20 ) ); assertThat( sum.sum(), is( 30L ) ); }
@Test void mustCombineWork() throws Exception { BinaryLatch startLatch = new BinaryLatch(); BinaryLatch blockLatch = new BinaryLatch(); FutureTask<Void> blocker = new FutureTask<>( new CallableWork( new AddWork( 1 ) { @Override public void apply( Adder adder ) { super.apply( adder ); startLatch.release(); blockLatch.await(); } } ) ); new Thread( blocker ).start(); startLatch.await(); Collection<FutureTask<Void>> tasks = new ArrayList<>(); tasks.add( blocker ); for ( int i = 0; i < 20; i++ ) { CallableWork task = new CallableWork( new AddWork( 1 ) ); FutureTask<Void> futureTask = new FutureTask<>( task ); tasks.add( futureTask ); Thread thread = new Thread( futureTask ); thread.start(); while ( thread.getState() != Thread.State.TIMED_WAITING ) { } } blockLatch.release(); for ( FutureTask<Void> task : tasks ) { task.get(); } assertThat( count.sum(), lessThan( sum.sum() ) ); }
@Test void mustApplyWorkEvenWhenInterrupted() throws Exception { Thread.currentThread().interrupt(); sync.apply( new AddWork( 10 ) ); assertThat( sum.sum(), is( 10L ) ); assertTrue( Thread.interrupted() ); }
@Test void mustRecoverFromExceptions() throws Exception { final AtomicBoolean broken = new AtomicBoolean( true ); Adder adder = new Adder() { @Override public void add( int delta ) { if ( broken.get() ) { throw new IllegalStateException( "boom!" ); } super.add( delta ); } }; sync = new WorkSync<>( adder ); try { executor.submit( new CallableWork( new AddWork( 10 ) ) ).get(); fail( "Should have thrown" ); } catch ( ExecutionException exception ) { assertThat( exception.getCause(), instanceOf( ExecutionException.class ) ); exception = (ExecutionException) exception.getCause(); assertThat( exception.getCause(), instanceOf( IllegalStateException.class ) ); } broken.set( false ); sync.apply( new AddWork( 20 ) ); assertThat( sum.sum(), is( 20L ) ); assertThat( count.sum(), is( 1L ) ); }
@Test void mustNotApplyWorkInParallelUnderStress() 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 ); for ( int i = 0; i < iterations; i++ ) { sync.apply( new AddWork( incrementValue ) ); } 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) ) ); }