@Test public void testIncrementPastIntegerMaxValue() { final long toTry = AcknowledgedCounterGenerator.WINDOW_SIZE * 3; AcknowledgedCounterGenerator generator = new AcknowledgedCounterGenerator(Integer.MAX_VALUE - 1000); Random rand = new Random(System.currentTimeMillis()); BlockingQueue<Long> pending = new ArrayBlockingQueue<Long>(1000); for (long i = 0; i < toTry; ++i) { long value = generator.nextValue(); while (!pending.offer(value)) { Long first = pending.poll(); if (rand.nextBoolean()) { generator.acknowledge(first); } else { Long second = pending.poll(); pending.add(first); generator.acknowledge(second); } } } }
public void acknowledge(long value) { final int currentSlot = (int)(value & WINDOW_MASK); if (window[currentSlot]) { throw new RuntimeException("Too many unacknowledged insertion keys."); } window[currentSlot] = true; if (lock.tryLock()) { try { long beforeFirstSlot = (limit & WINDOW_MASK); long index; for (index = limit + 1; index != beforeFirstSlot; ++index) { int slot = (int)(index & WINDOW_MASK); if (!window[slot]) { break; } window[slot] = false; } limit = index - 1; } finally { lock.unlock(); } } }
AcknowledgedCounterGenerator extends CounterGenerator { public void acknowledge(long value) { final int currentSlot = (int)(value & WINDOW_MASK); if (window[currentSlot]) { throw new RuntimeException("Too many unacknowledged insertion keys."); } window[currentSlot] = true; if (lock.tryLock()) { try { long beforeFirstSlot = (limit & WINDOW_MASK); long index; for (index = limit + 1; index != beforeFirstSlot; ++index) { int slot = (int)(index & WINDOW_MASK); if (!window[slot]) { break; } window[slot] = false; } limit = index - 1; } finally { lock.unlock(); } } } }
AcknowledgedCounterGenerator extends CounterGenerator { public void acknowledge(long value) { final int currentSlot = (int)(value & WINDOW_MASK); if (window[currentSlot]) { throw new RuntimeException("Too many unacknowledged insertion keys."); } window[currentSlot] = true; if (lock.tryLock()) { try { long beforeFirstSlot = (limit & WINDOW_MASK); long index; for (index = limit + 1; index != beforeFirstSlot; ++index) { int slot = (int)(index & WINDOW_MASK); if (!window[slot]) { break; } window[slot] = false; } limit = index - 1; } finally { lock.unlock(); } } } AcknowledgedCounterGenerator(long countstart); }
AcknowledgedCounterGenerator extends CounterGenerator { public void acknowledge(long value) { final int currentSlot = (int)(value & WINDOW_MASK); if (window[currentSlot]) { throw new RuntimeException("Too many unacknowledged insertion keys."); } window[currentSlot] = true; if (lock.tryLock()) { try { long beforeFirstSlot = (limit & WINDOW_MASK); long index; for (index = limit + 1; index != beforeFirstSlot; ++index) { int slot = (int)(index & WINDOW_MASK); if (!window[slot]) { break; } window[slot] = false; } limit = index - 1; } finally { lock.unlock(); } } } AcknowledgedCounterGenerator(long countstart); @Override Long lastValue(); void acknowledge(long value); }
AcknowledgedCounterGenerator extends CounterGenerator { public void acknowledge(long value) { final int currentSlot = (int)(value & WINDOW_MASK); if (window[currentSlot]) { throw new RuntimeException("Too many unacknowledged insertion keys."); } window[currentSlot] = true; if (lock.tryLock()) { try { long beforeFirstSlot = (limit & WINDOW_MASK); long index; for (index = limit + 1; index != beforeFirstSlot; ++index) { int slot = (int)(index & WINDOW_MASK); if (!window[slot]) { break; } window[slot] = false; } limit = index - 1; } finally { lock.unlock(); } } } AcknowledgedCounterGenerator(long countstart); @Override Long lastValue(); void acknowledge(long value);  }