[core] Update the AcknowledgedCounterGenerator to use a power of two

size for the window so a mask can be used instead of a modulo operation.
This avoids issues when there is overflow of the id integer values to
negative values.

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

+/**
+ * Copyright (c) 2015 YCSB contributors. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License. You
+ * may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License. See accompanying
+ * LICENSE file.
+ */
 package com.yahoo.ycsb.generator;
 
 import java.util.concurrent.locks.ReentrantLock;
@@ -8,7 +24,11 @@ import java.util.concurrent.locks.ReentrantLock;
  */
 public class AcknowledgedCounterGenerator extends CounterGenerator
 {
-	private static final int WINDOW_SIZE = 1000000;
+	/** The size of the window of pending id ack's. 2^20 = {@value} */
+	static final int WINDOW_SIZE = Integer.rotateLeft(1, 20);
+
+	/** The mask to use to turn an id into a slot in {@link #window}. */
+	private static final int WINDOW_MASK = WINDOW_SIZE - 1;
 
 	private final ReentrantLock lock;
 	private final boolean[] window;
@@ -40,22 +60,22 @@ public class AcknowledgedCounterGenerator extends CounterGenerator
 	 */
 	public void acknowledge(int value)
 	{
-		if (value > limit + WINDOW_SIZE) {
+		final int currentSlot = (value & WINDOW_MASK);
+		if (window[currentSlot] == true) {
 			throw new RuntimeException("Too many unacknowledged insertion keys.");
 		}
 
-		window[value % WINDOW_SIZE] = true;
+		window[currentSlot] = true;
 
 		if (lock.tryLock()) {
 			// move a contiguous sequence from the window
 			// over to the "limit" variable
-
 			try {
+			  // Only loop through the entire window at most once.
+			  int beforeFirstSlot = (limit & WINDOW_MASK);
 				int index;
-
-				for (index = limit + 1; index <= value; ++index) {
-					int slot = index % WINDOW_SIZE;
-
+				for (index = limit + 1; index != beforeFirstSlot; ++index) {
+					int slot = (index & WINDOW_MASK);
 					if (!window[slot]) {
 						break;
 					}

core/src/test/java/com/yahoo/ycsb/generator/AcknowledgedCounterGeneratorTest.java

+/**
+ * Copyright (c) 2015 YCSB contributors. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License. You
+ * may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License. See accompanying
+ * LICENSE file.
+ */
+package com.yahoo.ycsb.generator;
+
+import java.util.Random;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+
+import org.testng.annotations.Test;
+
+/**
+ * Tests for the AcknowledgedCounterGenerator class.
+ */
+public class AcknowledgedCounterGeneratorTest {
+
+  /**
+   * Test that advancing past {@link Integer#MAX_VALUE} works.
+   */
+  @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<Integer> pending = new ArrayBlockingQueue<Integer>(1000);
+    for (long i = 0; i < toTry; ++i) {
+      int value = generator.nextInt();
+
+      while (!pending.offer(value)) {
+
+        Integer first = pending.poll();
+
+        // Don't always advance by one.
+        if (rand.nextBoolean()) {
+          generator.acknowledge(first);
+        } else {
+          Integer second = pending.poll();
+          pending.add(first);
+          generator.acknowledge(second);
+        }
+      }
+    }
+
+  }
+}