From 5bb8596fabefb280a3f25e4e97c5859c1f0b3c1f Mon Sep 17 00:00:00 2001 From: gojomo Date: Tue, 28 Jul 2009 23:02:11 +0000 Subject: [PATCH] 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 --- .../archive/util/LongToIntConsistentHash.java | 105 +++++++++++++++ .../util/LongToIntConsistentHashTest.java | 54 ++++++++ .../HostnameQueueAssignmentPolicy.java | 100 +++++---------- .../frontier/QueueAssignmentPolicy.java | 43 +++---- .../SurtAuthorityQueueAssignmentPolicy.java | 101 ++++----------- ...RIAuthorityBasedQueueAssignmentPolicy.java | 120 ++++++++++++++++++ .../org/archive/modules/extractor/Hop.java | 40 +++--- 7 files changed, 375 insertions(+), 188 deletions(-) create mode 100644 commons/src/main/java/org/archive/util/LongToIntConsistentHash.java create mode 100644 commons/src/main/java/org/archive/util/LongToIntConsistentHashTest.java create mode 100644 engine/src/main/java/org/archive/crawler/frontier/URIAuthorityBasedQueueAssignmentPolicy.java diff --git a/commons/src/main/java/org/archive/util/LongToIntConsistentHash.java b/commons/src/main/java/org/archive/util/LongToIntConsistentHash.java new file mode 100644 index 00000000..f8064809 --- /dev/null +++ b/commons/src/main/java/org/archive/util/LongToIntConsistentHash.java @@ -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 circle = new TreeMap(); + 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 tailMap = circle.tailMap(longHash, true); + Map.Entry match = null; + for(Map.Entry 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); + } +} diff --git a/commons/src/main/java/org/archive/util/LongToIntConsistentHashTest.java b/commons/src/main/java/org/archive/util/LongToIntConsistentHashTest.java new file mode 100644 index 00000000..05a8f2e0 --- /dev/null +++ b/commons/src/main/java/org/archive/util/LongToIntConsistentHashTest.java @@ -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); + } +} diff --git a/engine/src/main/java/org/archive/crawler/frontier/HostnameQueueAssignmentPolicy.java b/engine/src/main/java/org/archive/crawler/frontier/HostnameQueueAssignmentPolicy.java index c3fd7ce7..bba3366f 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/HostnameQueueAssignmentPolicy.java +++ b/engine/src/main/java/org/archive/crawler/frontier/HostnameQueueAssignmentPolicy.java @@ -1,34 +1,26 @@ -/* HostnameQueueAssignmentPolicy -* -* $Id$ -* -* Created on Oct 5, 2004 -* -* Copyright (C) 2004 Internet Archive. -* -* This file is part of the Heritrix web crawler (crawler.archive.org). -* -* Heritrix is free software; you can redistribute it and/or modify -* it under the terms of the GNU Lesser Public License as published by -* the Free Software Foundation; either version 2.1 of the License, or -* any later version. -* -* Heritrix is distributed in the hope that it will be useful, -* but WITHOUT ANY WARRANTY; without even the implied warranty of -* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -* GNU Lesser Public License for more details. -* -* You should have received a copy of the GNU Lesser Public License -* along with Heritrix; if not, write to the Free Software -* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA -*/ +/* + * 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.crawler.frontier; -import java.util.logging.Level; -import java.util.logging.Logger; - import org.apache.commons.httpclient.URIException; -import org.archive.crawler.datamodel.CrawlURI; +import org.apache.commons.lang.StringUtils; import org.archive.net.UURI; import org.archive.net.UURIFactory; @@ -38,51 +30,22 @@ import org.archive.net.UURIFactory; * * @author gojomo */ -public class HostnameQueueAssignmentPolicy extends QueueAssignmentPolicy { - +public class HostnameQueueAssignmentPolicy +extends URIAuthorityBasedQueueAssignmentPolicy { private static final long serialVersionUID = 3L; - private static final Logger logger = Logger - .getLogger(HostnameQueueAssignmentPolicy.class.getName()); - /** - * When neat host-based class-key fails us - */ - private static String DEFAULT_CLASS_KEY = "default..."; - - private static final String DNS = "dns"; - - public String getClassKey(CrawlURI cauri) { - String scheme = cauri.getUURI().getScheme(); + @Override + String getCoreKey(UURI basis) { + String scheme = basis.getScheme(); String candidate = null; try { - if (scheme.equals(DNS)){ - if (cauri.getVia() != null) { - // Special handling for DNS: treat as being - // of the same class as the triggering URI. - // When a URI includes a port, this ensures - // the DNS lookup goes atop the host:port - // queue that triggered it, rather than - // some other host queue - UURI viaUuri = UURIFactory.getInstance(cauri.flattenVia()); - candidate = viaUuri.getAuthorityMinusUserinfo(); - // adopt scheme of triggering URI - scheme = viaUuri.getScheme(); - } else { - candidate= cauri.getUURI().getReferencedHost(); - } - } else { - candidate = cauri.getUURI().getAuthorityMinusUserinfo(); - } - - if(candidate == null || candidate.length() == 0) { - candidate = DEFAULT_CLASS_KEY; - } - } catch (URIException e) { - logger.log(Level.INFO, - "unable to extract class key; using default", e); - candidate = DEFAULT_CLASS_KEY; + candidate = basis.getAuthorityMinusUserinfo(); + } catch (URIException ue) {}// let next line handle + + if(StringUtils.isEmpty(candidate)) { + return null; } - if (scheme != null && scheme.equals(UURIFactory.HTTPS)) { + if (UURIFactory.HTTPS.equals(scheme)) { // If https and no port specified, add default https port to // distinguish https from http server without a port. if (!candidate.matches(".+:[0-9]+")) { @@ -93,4 +56,5 @@ public class HostnameQueueAssignmentPolicy extends QueueAssignmentPolicy { return candidate.replace(':','#'); } + } diff --git a/engine/src/main/java/org/archive/crawler/frontier/QueueAssignmentPolicy.java b/engine/src/main/java/org/archive/crawler/frontier/QueueAssignmentPolicy.java index 2afe04ab..7c1a4bae 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/QueueAssignmentPolicy.java +++ b/engine/src/main/java/org/archive/crawler/frontier/QueueAssignmentPolicy.java @@ -1,27 +1,22 @@ -/* QueueAssignmentPolicy -* -* $Id$ -* -* Created on Oct 5, 2004 -* -* Copyright (C) 2004 Internet Archive. -* -* This file is part of the Heritrix web crawler (crawler.archive.org). -* -* Heritrix is free software; you can redistribute it and/or modify -* it under the terms of the GNU Lesser Public License as published by -* the Free Software Foundation; either version 2.1 of the License, or -* any later version. -* -* Heritrix is distributed in the hope that it will be useful, -* but WITHOUT ANY WARRANTY; without even the implied warranty of -* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -* GNU Lesser Public License for more details. -* -* You should have received a copy of the GNU Lesser Public License -* along with Heritrix; if not, write to the Free Software -* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA -*/ +/* + * 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.crawler.frontier; import java.io.Serializable; diff --git a/engine/src/main/java/org/archive/crawler/frontier/SurtAuthorityQueueAssignmentPolicy.java b/engine/src/main/java/org/archive/crawler/frontier/SurtAuthorityQueueAssignmentPolicy.java index 1e2243a1..c19a3183 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/SurtAuthorityQueueAssignmentPolicy.java +++ b/engine/src/main/java/org/archive/crawler/frontier/SurtAuthorityQueueAssignmentPolicy.java @@ -1,92 +1,39 @@ -/* SurtAuthorityQueueAssignmentPolicy -* -* $Id$ -* -* Created on Oct 5, 2004 -* -* Copyright (C) 2004 Internet Archive. -* -* This file is part of the Heritrix web crawler (crawler.archive.org). -* -* Heritrix is free software; you can redistribute it and/or modify -* it under the terms of the GNU Lesser Public License as published by -* the Free Software Foundation; either version 2.1 of the License, or -* any later version. -* -* Heritrix is distributed in the hope that it will be useful, -* but WITHOUT ANY WARRANTY; without even the implied warranty of -* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -* GNU Lesser Public License for more details. -* -* You should have received a copy of the GNU Lesser Public License -* along with Heritrix; if not, write to the Free Software -* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA -*/ +/* + * 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.crawler.frontier; -import java.util.logging.Level; -import java.util.logging.Logger; - -import org.apache.commons.httpclient.URIException; -import org.archive.crawler.datamodel.CrawlURI; import org.archive.net.UURI; -import org.archive.net.UURIFactory; /** * SurtAuthorityQueueAssignmentPolicy based on the surt form of hostname. */ public class SurtAuthorityQueueAssignmentPolicy -extends QueueAssignmentPolicy { - +extends URIAuthorityBasedQueueAssignmentPolicy { private static final long serialVersionUID = 3L; - - private static final Logger logger = Logger - .getLogger(SurtAuthorityQueueAssignmentPolicy.class.getName()); - /** - * When neat host-based class-key fails us - */ - private static String DEFAULT_CLASS_KEY = "default..."; - - private static final String DNS = "dns"; - - public String getClassKey(CrawlURI cauri) { - String scheme = cauri.getUURI().getScheme(); - String candidate = null; - try { - if (scheme.equals(DNS)) { - UURI effectiveuuri; - if (cauri.getVia() != null) { - // Special handling for DNS: treat as being - // of the same class as the triggering URI. - // When a URI includes a port, this ensures - // the DNS lookup goes atop the host:port - // queue that triggered it, rather than - // some other host queue - effectiveuuri = UURIFactory.getInstance(cauri.flattenVia()); - } else { - // To get the dns surt form, create a fake http version - // (Gordon suggestion). - effectiveuuri = UURIFactory.getInstance("http://" + - cauri.getUURI().getPath()); - } - candidate = getSurtAuthority(effectiveuuri.getSurtForm()); - } else { - candidate = getSurtAuthority(cauri.getUURI().getSurtForm()); - } - - if(candidate == null || candidate.length() == 0) { - candidate = DEFAULT_CLASS_KEY; - } - } catch (URIException e) { - logger.log(Level.INFO, - "unable to extract class key; using default", e); - candidate = DEFAULT_CLASS_KEY; - } - // Ensure classKeys are safe as filenames on NTFS + @Override + String getCoreKey(UURI basis) { + String candidate = getSurtAuthority(basis.getSurtForm()); return candidate.replace(':','#'); } - + protected String getSurtAuthority(String surt) { int indexOfOpen = surt.indexOf("://("); int indexOfClose = surt.indexOf(")"); diff --git a/engine/src/main/java/org/archive/crawler/frontier/URIAuthorityBasedQueueAssignmentPolicy.java b/engine/src/main/java/org/archive/crawler/frontier/URIAuthorityBasedQueueAssignmentPolicy.java new file mode 100644 index 00000000..c2c0d351 --- /dev/null +++ b/engine/src/main/java/org/archive/crawler/frontier/URIAuthorityBasedQueueAssignmentPolicy.java @@ -0,0 +1,120 @@ +/* + * 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.crawler.frontier; + +import org.apache.commons.lang.StringUtils; +import org.archive.crawler.datamodel.CrawlURI; +import org.archive.modules.extractor.Hop; +import org.archive.net.UURI; +import org.archive.spring.HasKeyedProperties; +import org.archive.spring.KeyedProperties; +import org.archive.util.LongToIntConsistentHash; + +/** + * SurtAuthorityQueueAssignmentPolicy based on the surt form of hostname. + */ +public abstract class URIAuthorityBasedQueueAssignmentPolicy +extends + QueueAssignmentPolicy +implements + HasKeyedProperties { + private static final long serialVersionUID = 3L; + + KeyedProperties kp = new KeyedProperties(); + public KeyedProperties getKeyedProperties() { + return kp; + } + //for when neat class-key fails us + protected static String DEFAULT_CLASS_KEY = "default..."; + + LongToIntConsistentHash conhash = new LongToIntConsistentHash(); + + /** + * Whether to always defer to a previously-assigned key inside + * the CrawlURI. If true, any key already in the CrawlURI will + * be returned as the classKey. + */ + public boolean getDeferToPrevious() { + return (Boolean) kp.get("deferToPrevious"); + } + { + setDeferToPrevious(true); + } + public void setDeferToPrevious(boolean defer) { + kp.put("deferToPrevious",defer); + } + + /** + * The number of parallel queues to split a core key into. By + * default is 1. If larger than 1, the non-authority-based portion + * of the URI will be used to distribute over that many separate + * queues. + * + */ + public int getParallelQueues() { + return (Integer) kp.get("parallelQueues"); + } + { + setParallelQueues(1); + } + public void setParallelQueues(int count) { + kp.put("parallelQueues",count); + } + + public String getClassKey(CrawlURI curi) { + if(getDeferToPrevious() && !StringUtils.isEmpty(curi.getClassKey())) { + return curi.getClassKey(); + } + + UURI basis = getBasisURI(curi); + String candidate = getCoreKey(basis); + + if(StringUtils.isEmpty(candidate)) { + return DEFAULT_CLASS_KEY; + } + + if(getParallelQueues()>1) { + int subqueue = getSubqueue(basis,getParallelQueues()); + if (subqueue>0) { + candidate += "+"+subqueue; + } + } + return candidate; + } + + protected int getSubqueue(UURI basis, int parallelQueues) { + return conhash.bucketFor(basis.getRawPathQuery(), parallelQueues); + } + + abstract String getCoreKey(UURI basis); + + protected UURI getBasisURI(CrawlURI curi) { + UURI effectiveuuri = null; + // always use 'via' of prerequisite URIs, if available, so + // prerequisites go to same queue as trigger URI + if (curi.getPathFromSeed().endsWith(Hop.PREREQ.getHopString())) { + effectiveuuri = curi.getVia(); + } + if(effectiveuuri==null) { + effectiveuuri = curi.getUURI(); + } + return effectiveuuri; + } +} diff --git a/modules/src/main/java/org/archive/modules/extractor/Hop.java b/modules/src/main/java/org/archive/modules/extractor/Hop.java index dfbe8de0..775bdb97 100644 --- a/modules/src/main/java/org/archive/modules/extractor/Hop.java +++ b/modules/src/main/java/org/archive/modules/extractor/Hop.java @@ -1,26 +1,22 @@ -/* Copyright (C) 2006 Internet Archive. +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). * - * This file is part of the Heritrix web crawler (crawler.archive.org). + * Licensed to the Internet Archive (IA) by one or more individual + * contributors. * - * Heritrix is free software; you can redistribute it and/or modify - * it under the terms of the GNU Lesser Public License as published by - * the Free Software Foundation; either version 2.1 of the License, or - * any later version. + * 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 * - * Heritrix is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Lesser Public License for more details. + * http://www.apache.org/licenses/LICENSE-2.0 * - * You should have received a copy of the GNU Lesser Public License - * along with Heritrix; if not, write to the Free Software - * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA - * - * Hop.java - * Created on October 5, 2006 - * - * $Header$ + * 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.modules.extractor; @@ -57,7 +53,7 @@ public enum Hop { /** The hop character for logs. */ private char hopChar; - + protected String hopString; /** * Constructor. @@ -66,6 +62,7 @@ public enum Hop { */ private Hop(char hopChar) { this.hopChar = hopChar; + this.hopString = ""+hopChar; } @@ -77,4 +74,9 @@ public enum Hop { public char getHopChar() { return hopChar; } + + + public String getHopString() { + return hopString; + } }