Work for [HER-1656] add multiple-queues-per-host ('parallelQueues') capabilities to QueueAssignmentPolicies

* LongToIntConsistentHash.java + Test
    expand range/distribution tests
    correct problems with phony '-1' bucket being created/returned

commons/src/main/java/org/archive/util/LongToIntConsistentHash.java

@@ -31,37 +31,57 @@ import st.ata.util.FPGenerator;
  * integer. 
  */
 public class LongToIntConsistentHash {
+    protected static final int DEFAULT_REPLICAS = 128;
     TreeMap<Long,Integer> circle = new TreeMap<Long,Integer>();
     int replicasInstalledUpTo=-1; 
-    int numReplicas = 32; 
+    int numReplicas; 
 
     public LongToIntConsistentHash() {
-        this(32); 
+        this(DEFAULT_REPLICAS); 
     }
     
     public LongToIntConsistentHash(int numReplicas) {
         this.numReplicas = numReplicas;
         installReplicas(0);
+        replicasInstalledUpTo=1;
     }
 
     /**
      * Install necessary replicas, if not already present.
      * @param upTo
      */
-    public void installReplicas(int upTo) {
+    public void installReplicasUpTo(int upTo) {
         if(replicasInstalledUpTo>upTo) {
             return;
         }
         for(;replicasInstalledUpTo<upTo;replicasInstalledUpTo++) {
-            for(int i = 0; i < numReplicas; i++) {
-                circle.put(
-                        FPGenerator.std64.fp(
-                                ""+replicasInstalledUpTo+":"+i),
-                        replicasInstalledUpTo);
-            }
+            installReplicas(replicasInstalledUpTo);
+        }
+    }
+
+    private void installReplicas(int bucket) {
+        for(int i = 0; i < numReplicas; i++) {
+            circle.put(
+                    replicaLocation(bucket,i),
+                    bucket);
         }
     }
     
+//    SecureRandom rand = new SecureRandom(); 
+    protected long replicaLocation(int bucketNumber, int replicaNumber) {
+//      return rand.nextLong();
+//      return RandomUtils.nextLong();
+//      return ArchiveUtils.doubleMurmur(string.getBytes());
+//      return (new JenkinsHash()).hash(string.getBytes());
+       return FPGenerator.std64.fp(bucketNumber+"."+replicaNumber);
+    }
+    
+    protected long hash(String string) {
+//      return ArchiveUtils.doubleMurmur(string.getBytes());
+//      return (new JenkinsHash()).hash(string.getBytes());
+       return FPGenerator.std64.fp(string);
+    }
+
     /**
      * Return the proper integer bucket-number for the given long hash,
      * up to the given integer boundary (exclusive). 
@@ -71,7 +91,7 @@ public class LongToIntConsistentHash {
      * @return
      */
     public int bucketFor(long longHash, int upTo) {
-        installReplicas(upTo); 
+        installReplicasUpTo(upTo); 
         
         NavigableMap<Long, Integer> tailMap = circle.tailMap(longHash, true);
         Map.Entry<Long,Integer> match = null;
@@ -96,10 +116,10 @@ public class LongToIntConsistentHash {
      * @return
      */
     public int bucketFor(CharSequence cs, int upTo) {
-        return bucketFor(FPGenerator.std64.fp(cs), upTo);
+        return bucketFor(hash(cs.toString()), upTo);
     }
 
     public int bucketFor(char[] chars, int upTo) {
-        return bucketFor(FPGenerator.std64.fp(chars,0,chars.length), upTo);
+        return bucketFor(hash(new String(chars)), upTo);
     }
 }

commons/src/main/java/org/archive/util/LongToIntConsistentHashTest.java

@@ -1,3 +1,22 @@
+/*
+ *  This file is part of the Heritrix web crawler (crawler.archive.org).
+ *
+ *  Licensed to the Internet Archive (IA) by one or more individual 
+ *  contributors. 
+ *
+ *  The IA licenses this file to You 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.
+ */
+
 package org.archive.util;
 
 import junit.framework.TestCase;
@@ -21,6 +40,23 @@ public class LongToIntConsistentHashTest extends TestCase {
             int upTo = RandomUtils.nextInt(32)+1;
             int bucket = conhash.bucketFor(longHash, upTo);
             assertTrue("bucket returned >= upTo",bucket < upTo);
+            assertTrue("bucket returned < 0: "+bucket,bucket >= 0);
+
+        }
+    }
+    
+    public void testTwoWayDistribution() {
+//        SecureRandom rand = new SecureRandom("foobar".getBytes()); 
+        for(int p = 0; p < 20; p++) {
+            int[] landings = new int[2];
+            for(long in = 0; in < 100000; in++) {
+                long longHash = FPGenerator.std64.fp(p+"a"+in);
+//                long longHash = rand.nextLong();
+//                long longHash = ArchiveUtils.doubleMurmur((p+":"+in).getBytes());
+                landings[conhash.bucketFor(longHash, 2)]++;
+            }
+//            System.out.println(landings[0]+","+landings[1]);
+            assertTrue("excessive changes",Math.abs(landings[0]-landings[1]) < 2000); 
         }
     }
     
@@ -35,7 +71,7 @@ public class LongToIntConsistentHashTest extends TestCase {
                 changedCount++;
             }
         }
-        assertTrue("excessive changes",changedCount < 2000); 
+        assertTrue("excessive changes: "+changedCount,changedCount < 2000); 
     }
     
     public void testConsistencyDown() {
@@ -49,6 +85,6 @@ public class LongToIntConsistentHashTest extends TestCase {
                 changedCount++;
             }
         }
-        assertTrue("excessive changes",changedCount < 2000); 
+        assertTrue("excessive changes: "+changedCount,changedCount < 2000); 
     }
 }