LearnerSnapshotThrottler { public LearnerSnapshot beginSnapshot(boolean essential) throws SnapshotThrottleException, InterruptedException { int snapshotNumber; synchronized (snapCountSyncObject) { if (!essential && timeoutMillis > 0 && snapsInProgress >= maxConcurrentSnapshots) { long timestamp = Time.currentElapsedTime(); do { snapCountSyncObject.wait(timeoutMillis); } while (snapsInProgress >= maxConcurrentSnapshots && timestamp + timeoutMillis < Time.currentElapsedTime()); } if (essential || snapsInProgress < maxConcurrentSnapshots) { snapsInProgress++; snapshotNumber = snapsInProgress; } else { throw new SnapshotThrottleException(snapsInProgress + 1, maxConcurrentSnapshots); } } return new LearnerSnapshot(this, snapshotNumber, essential); } LearnerSnapshotThrottler(int maxConcurrentSnapshots,
                                    long timeoutMillis);  LearnerSnapshotThrottler(int maxConcurrentSnapshots); LearnerSnapshot beginSnapshot(boolean essential); void endSnapshot();  }
@Test(expected = SnapshotThrottleException.class) public void testTooManySnapshotsNonessential() throws Exception { LearnerSnapshotThrottler throttler = new LearnerSnapshotThrottler(5); for (int i = 0; i < 6; i++) { throttler.beginSnapshot(false); } }
@Test public void testTryWithResourceNoThrottle() throws Exception { LearnerSnapshotThrottler throttler = new LearnerSnapshotThrottler(1); for (int i = 0; i < 3; i++) { LearnerSnapshot snapshot = throttler.beginSnapshot(false); try { Assert.assertFalse(snapshot.isEssential()); Assert.assertEquals(1, snapshot.getConcurrentSnapshotNumber()); } finally { snapshot.close(); } } }
@Test(expected = SnapshotThrottleException.class) public void testTryWithResourceThrottle() throws Exception { LearnerSnapshotThrottler throttler = new LearnerSnapshotThrottler(1); LearnerSnapshot outer = throttler.beginSnapshot(true); try { LearnerSnapshot inner = throttler.beginSnapshot(false); try { Assert.fail("shouldn't be able to have both snapshots open"); } finally { inner.close(); } } finally { outer.close(); } }
@Test public void testPositiveTimeout() throws Exception { final LearnerSnapshotThrottler throttler = new LearnerSnapshotThrottler(1, 200); ExecutorService threadPool = Executors.newFixedThreadPool(1); LearnerSnapshot first = throttler.beginSnapshot(false); final CountDownLatch snapshotProgressLatch = new CountDownLatch(1); Future<Boolean> result = threadPool.submit(new Callable<Boolean>() { @Override public Boolean call() { try { snapshotProgressLatch.countDown(); LearnerSnapshot second = throttler.beginSnapshot(false); second.close(); } catch (Exception e) { return false; } return true; } }); snapshotProgressLatch.await(); first.close(); Assert.assertTrue(result.get()); }