Merge pull request #364 from tlopatic/insert-fix

Proposed fix for #327.

core/src/main/java/com/yahoo/ycsb/generator/AcknowledgedCounterGenerator.java

@@ -0,0 +1,72 @@
+package com.yahoo.ycsb.generator;
+
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * A CounterGenerator that reports generated integers via lastInt()
+ * only after they have been acknowledged.
+ */
+public class AcknowledgedCounterGenerator extends CounterGenerator
+{
+	private static final int WINDOW_SIZE = 1000000;
+
+	private final ReentrantLock lock;
+	private final boolean[] window;
+	private volatile int limit;
+
+	/**
+	 * Create a counter that starts at countstart.
+	 */
+	public AcknowledgedCounterGenerator(int countstart)
+	{
+		super(countstart);
+		lock = new ReentrantLock();
+		window = new boolean[WINDOW_SIZE];
+		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 void acknowledge(int value)
+	{
+		if (value > limit + WINDOW_SIZE) {
+			throw new RuntimeException("Too many unacknowledged insertion keys.");
+		}
+
+		window[value % WINDOW_SIZE] = true;
+
+		if (lock.tryLock()) {
+			// move a contiguous sequence from the window
+			// over to the "limit" variable
+
+			try {
+				int index;
+
+				for (index = limit + 1; index <= value; ++index) {
+					int slot = index % WINDOW_SIZE;
+
+					if (!window[slot]) {
+						break;
+					}
+
+					window[slot] = false;
+				}
+
+				limit = index - 1;
+			} finally {
+				lock.unlock();
+			}
+		}
+	}
+}

core/src/main/java/com/yahoo/ycsb/workloads/CoreWorkload.java

@@ -20,6 +20,7 @@
 import java.util.Properties;
 
 import com.yahoo.ycsb.*;
+import com.yahoo.ycsb.generator.AcknowledgedCounterGenerator;
 import com.yahoo.ycsb.generator.CounterGenerator;
 import com.yahoo.ycsb.generator.DiscreteGenerator;
 import com.yahoo.ycsb.generator.ExponentialGenerator;
@@ -299,7 +300,7 @@ public class CoreWorkload extends Workload
 
 	Generator fieldchooser;
 
-	CounterGenerator transactioninsertkeysequence;
+	AcknowledgedCounterGenerator transactioninsertkeysequence;
 
 	IntegerGenerator scanlength;
 
@@ -417,7 +418,7 @@ else if (requestdistrib.compareTo("exponential")==0)
 			operationchooser.addValue(readmodifywriteproportion,"READMODIFYWRITE");
 		}
 
-		transactioninsertkeysequence=new CounterGenerator(recordcount);
+		transactioninsertkeysequence=new AcknowledgedCounterGenerator(recordcount);
 		if (requestdistrib.compareTo("uniform")==0)
 		{
 			keychooser=new UniformIntegerGenerator(0,recordcount-1);
@@ -759,9 +760,13 @@ public void doTransactionInsert(DB db)
 		//choose the next key
 		int keynum=transactioninsertkeysequence.nextInt();
 
-		String dbkey = buildKeyName(keynum);
+		try {
+			String dbkey = buildKeyName(keynum);
 
-		HashMap<String, ByteIterator> values = buildValues(dbkey);
-		db.insert(table,dbkey,values);
+			HashMap<String, ByteIterator> values = buildValues(dbkey);
+			db.insert(table,dbkey,values);
+		} finally {
+			transactioninsertkeysequence.acknowledge(keynum);
+		}
 	}
 }