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

* LongToIntConsistentHash.java + Test
    consistent-hashing utility class
* URIAuthorityBasedQueueAssignmentPolicy.java
    shared superclass for the hostname and surtauthority QAPs
    setting for 'deferToPrevious' -- avoid changing assignments
    setting for 'parallelQueues' -- when > 1, consistent-hash URIs across that many separate queues (with numerical suffix)
* SurtAuthorityQueueAssignmentPolicy.java, HostnameQueueAssignmentPolicy.java
    refactor to derive from URIAuthorityBasedQueueAssignmentPolicy
* QueueAssignmentPolicy.java
    add Apache license
* Hop.java
    add String representation for convenience
This commit is contained in:
gojomo committed 2009-07-28 23:02:11 +00:00
1 parent d2a82228a3
commit 5bb8596fab
7 files changed
+375 -188

No files matched your search

@@ -0,0 +1,105 @@
/*
* 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 java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;
import st.ata.util.FPGenerator;
/**
* Simple consistent-hashing implementation: provided a long and an
* integer bucket-number upper-bound (exclusive), return the matching
* integer.
*/
public class LongToIntConsistentHash {
TreeMap<Long,Integer> circle = new TreeMap<Long,Integer>();
int replicasInstalledUpTo=-1;
int numReplicas = 32;
public LongToIntConsistentHash() {
this(32);
}
public LongToIntConsistentHash(int numReplicas) {
this.numReplicas = numReplicas;
installReplicas(0);
}
/**
* Install necessary replicas, if not already present.
* @param upTo
*/
public void installReplicas(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);
}
}
}
/**
* Return the proper integer bucket-number for the given long hash,
* up to the given integer boundary (exclusive).
*
* @param longHash
* @param upTo
* @return
*/
public int bucketFor(long longHash, int upTo) {
installReplicas(upTo);
NavigableMap<Long, Integer> tailMap = circle.tailMap(longHash, true);
Map.Entry<Long,Integer> match = null;
for(Map.Entry<Long,Integer> candidate : tailMap.entrySet()) {
if(candidate.getValue() < upTo) {
match = candidate;
break;
}
}
if (match == null) {
return bucketFor(Long.MIN_VALUE,upTo);
}
return match.getValue();
}
/**
* Convenience alternative which creates longHash from CharSequence
*
* @param string
* @param upTo
* @return
*/
public int bucketFor(CharSequence cs, int upTo) {
return bucketFor(FPGenerator.std64.fp(cs), upTo);
}
public int bucketFor(char[] chars, int upTo) {
return bucketFor(FPGenerator.std64.fp(chars,0,chars.length), upTo);
}
}
@@ -0,0 +1,54 @@
package org.archive.util;
import junit.framework.TestCase;
import org.apache.commons.lang.math.RandomUtils;
import st.ata.util.FPGenerator;
public class LongToIntConsistentHashTest extends TestCase {
LongToIntConsistentHash conhash;
@Override
protected void setUp() throws Exception {
super.setUp();
conhash = new LongToIntConsistentHash();
}
public void testRange() {
for(long in = 0; in < 10000; in++) {
long longHash = FPGenerator.std64.fp(""+in);
int upTo = RandomUtils.nextInt(32)+1;
int bucket = conhash.bucketFor(longHash, upTo);
assertTrue("bucket returned >= upTo",bucket < upTo);
}
}
public void testConsistencyUp() {
int initialUpTo = 10;
int changedCount = 0;
for(long in = 0; in < 10000; in++) {
long longHash = FPGenerator.std64.fp(""+in);
int firstBucket = conhash.bucketFor(longHash, initialUpTo);
int secondBucket = conhash.bucketFor(longHash, initialUpTo+1);
if(secondBucket!=firstBucket) {
changedCount++;
}
}
assertTrue("excessive changes",changedCount < 2000);
}
public void testConsistencyDown() {
int initialUpTo = 10;
int changedCount = 0;
for(long in = 0; in < 10000; in++) {
long longHash = FPGenerator.std64.fp(""+in);
int firstBucket = conhash.bucketFor(longHash, initialUpTo);
int secondBucket = conhash.bucketFor(longHash, initialUpTo-1);
if(secondBucket!=firstBucket) {
changedCount++;
}
}
assertTrue("excessive changes",changedCount < 2000);
}
}