package com.yahoo.ycsb.generator;

import java.util.PriorityQueue;

/**
 * A CounterGenerator that reports generated integers via lastInt()
 * only after they have been acknowledged.
 */
public class AcknowledgedCounterGenerator extends CounterGenerator
{
	private PriorityQueue<Integer> ack;
	private int limit;

	/**
	 * Create a counter that starts at countstart.
	 */
	public AcknowledgedCounterGenerator(int countstart)
	{
		super(countstart);
		ack = new PriorityQueue<Integer>();
		limit = countstart - 1;
	}

	/**
	 * In this generator, the highest acknowledged counter value
	 * (as opposed to the highest generated counter value).
	 */
	@Override
	public int lastInt()
	{
		return limit;
	}

	/**
	 * Make a generated counter value available via lastInt().
	 */
	public synchronized void acknowledge(int value)
	{
		ack.add(value);

		// move a contiguous sequence from the priority queue
		// over to the "limit" variable

		Integer min;

		while ((min = ack.peek()) != null && min == limit + 1) {
			limit = ack.poll();
		}
	}
}

@@ -45,29 +45,46 @@
 va.util.
-PriorityQueue
+concurrent.locks.ReentrantLock
 ;%0A%0A/**%0A 
@@ -279,34 +279,101 @@
 ate 
-PriorityQueue%3CInteger%3E ack
+static final int WINDOW_SIZE = 10000;%0A%0A%09private ReentrantLock lock;%0A%09private boolean%5B%5D window
 ;%0A%09p
@@ -528,17 +528,18 @@
 art);%0A%09%09
-a
+lo
 ck = new
@@ -543,32 +543,60 @@
 new 
-PriorityQueue%3CInteger%3E()
+ReentrantLock();%0A%09%09window = new boolean%5BWINDOW_SIZE%5D
 ;%0A%09%09
@@ -890,21 +890,8 @@
 lic 
-synchronized 
 void
@@ -923,25 +923,184 @@
 %7B%0A%09%09
-ack.add(value);%0A%0A
+if (value %3E limit + WINDOW_SIZE) %7B%0A%09%09%09throw new RuntimeException(%22This should be a different exception.%22);%0A%09%09%7D%0A%0A%09%09window%5Bvalue %25 WINDOW_SIZE%5D = true;%0A%0A%09%09if (lock.tryLock()) %7B%0A%09
 %09%09//
@@ -1140,23 +1140,16 @@
 the 
-priority queue%0A
+window%0A%09
 %09%09//
@@ -1185,103 +1185,266 @@
 %0A%0A%09%09
-Integer min;%0A%0A%09%09while ((min = ack.peek()) != null && min == limit + 1) %7B%0A%09%09%09limit = ack.poll();
+%09try %7B%0A%09%09%09%09int index;%0A%0A%09%09%09%09for (index = limit + 1; index %3C= value; ++index) %7B%0A%09%09%09%09%09int slot = index %25 WINDOW_SIZE;%0A%0A%09%09%09%09%09if (!window%5Bslot%5D) %7B%0A%09%09%09%09%09%09break;%0A%09%09%09%09%09%7D%0A%0A%09%09%09%09%09window%5Bslot%5D = false;%0A%09%09%09%09%7D%0A%0A%09%09%09%09limit = index - 1;%0A%09%09%09%7D finally %7B%0A%09%09%09%09lock.unlock();%0A%09%09%09%7D
 %0A%09%09%7D

Use a sliding window to track acknowledged values.