@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()); }
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 { 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 { 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); }
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(); }
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();  }