mirror of
https://github.com/yacy/yacy_search_server.git
synced 2024-09-19 00:01:41 +02:00
561 lines
24 KiB
Java
561 lines
24 KiB
Java
/**
|
|
* HostQueues
|
|
* Copyright 2013 by Michael Christen
|
|
* First released 24.09.2013 at http://yacy.net
|
|
*
|
|
* This library is free software; you can redistribute it and/or
|
|
* modify it under the terms of the GNU Lesser General Public
|
|
* License as published by the Free Software Foundation; either
|
|
* version 2.1 of the License, or (at your option) any later version.
|
|
*
|
|
* This library 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 General Public License for more details.
|
|
*
|
|
* You should have received a copy of the GNU Lesser General Public License
|
|
* along with this program in the file lgpl21.txt
|
|
* If not, see <http://www.gnu.org/licenses/>.
|
|
*/
|
|
|
|
package net.yacy.crawler;
|
|
|
|
import java.io.File;
|
|
import java.io.IOException;
|
|
import java.lang.reflect.Array;
|
|
import java.net.MalformedURLException;
|
|
import java.util.ArrayList;
|
|
import java.util.Collections;
|
|
import java.util.ConcurrentModificationException;
|
|
import java.util.HashSet;
|
|
import java.util.Iterator;
|
|
import java.util.List;
|
|
import java.util.Map;
|
|
import java.util.Set;
|
|
import java.util.TreeMap;
|
|
import java.util.concurrent.ConcurrentHashMap;
|
|
|
|
import net.yacy.cora.document.encoding.ASCII;
|
|
import net.yacy.cora.document.id.DigestURL;
|
|
import net.yacy.cora.order.Base64Order;
|
|
import net.yacy.cora.protocol.ClientIdentification;
|
|
import net.yacy.cora.storage.HandleMap;
|
|
import net.yacy.cora.storage.HandleSet;
|
|
import net.yacy.cora.util.ConcurrentLog;
|
|
import net.yacy.cora.util.SpaceExceededException;
|
|
import net.yacy.crawler.data.CrawlProfile;
|
|
import net.yacy.crawler.data.Latency;
|
|
import net.yacy.crawler.retrieval.Request;
|
|
import net.yacy.crawler.robots.RobotsTxt;
|
|
import net.yacy.kelondro.data.word.Word;
|
|
import net.yacy.kelondro.index.RowHandleMap;
|
|
import net.yacy.kelondro.index.RowHandleSet;
|
|
import net.yacy.kelondro.util.FileUtils;
|
|
|
|
/**
|
|
* wrapper for single HostQueue queues; this is a collection of such queues.
|
|
* All these queues are stored in a common directory for the queue stacks.
|
|
*
|
|
* ATTENTION: the order of urls returned by this balancer must strictly follow the clickdepth order.
|
|
* That means that all links from a given host must be returned from the lowest crawldepth only.
|
|
* The crawldepth is interpreted as clickdepth and the crawler is producing that semantic using a
|
|
* correct crawl ordering.
|
|
*/
|
|
public class HostBalancer implements Balancer {
|
|
|
|
private final static ConcurrentLog log = new ConcurrentLog("HostBalancer");
|
|
public final static HandleMap depthCache = new RowHandleMap(Word.commonHashLength, Word.commonHashOrder, 2, 8 * 1024 * 1024, "HostBalancer.DepthCache");
|
|
|
|
private final File hostsPath;
|
|
private final boolean exceed134217727;
|
|
private final Map<String, HostQueue> queues;
|
|
private final Set<String> roundRobinHostHashes;
|
|
private final int onDemandLimit;
|
|
|
|
/**
|
|
* Create a new instance and asynchronously fills the queue by scanning the hostsPath directory.
|
|
* @param hostsPath path with persisted hosts queues
|
|
* @param onDemandLimit
|
|
* @param exceed134217727
|
|
*/
|
|
public HostBalancer(
|
|
final File hostsPath,
|
|
final int onDemandLimit,
|
|
final boolean exceed134217727) {
|
|
this(hostsPath, onDemandLimit, exceed134217727, true);
|
|
}
|
|
|
|
/**
|
|
* Create a new instance and fills the queue by scanning the hostsPath directory.
|
|
* @param hostsPath
|
|
* @param onDemandLimit
|
|
* @param exceed134217727
|
|
* @param asyncInit when true, queue filling from file system is launched asynchronously
|
|
*/
|
|
public HostBalancer(
|
|
final File hostsPath,
|
|
final int onDemandLimit,
|
|
final boolean exceed134217727,
|
|
final boolean asyncInit) {
|
|
this.hostsPath = hostsPath;
|
|
this.onDemandLimit = onDemandLimit;
|
|
this.exceed134217727 = exceed134217727;
|
|
|
|
// create a stack for newly entered entries
|
|
if (!(hostsPath.exists())) hostsPath.mkdirs(); // make the path
|
|
this.queues = new ConcurrentHashMap<String, HostQueue>();
|
|
this.roundRobinHostHashes = new HashSet<String>();
|
|
init(asyncInit); // return without wait but starts a thread to fill the queues
|
|
}
|
|
|
|
/**
|
|
* Fills the queue by scanning the hostsPath directory.
|
|
* @param async when true, launch in a dedicated thread to
|
|
* return immediately (as large unfinished crawls may take longer to load)
|
|
*/
|
|
private void init(final boolean async) {
|
|
if(async) {
|
|
Thread t = new Thread("HostBalancer.init") {
|
|
@Override
|
|
public void run() {
|
|
runInit();
|
|
}
|
|
};
|
|
|
|
t.start();
|
|
} else {
|
|
runInit();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Fills the queue by scanning the hostsPath directory.
|
|
*/
|
|
private void runInit() {
|
|
final String[] hostlist = hostsPath.list();
|
|
for (String hoststr : hostlist) {
|
|
try {
|
|
File queuePath = new File(hostsPath, hoststr);
|
|
HostQueue queue = new HostQueue(queuePath, queues.size() > onDemandLimit, exceed134217727);
|
|
if (queue.isEmpty()) {
|
|
queue.close();
|
|
FileUtils.deletedelete(queuePath);
|
|
} else {
|
|
queues.put(queue.getHostHash(), queue);
|
|
}
|
|
} catch (MalformedURLException | RuntimeException e) {
|
|
log.warn("delete queue due to init error for " + hostsPath.getName() + " host=" + hoststr + " " + e.getLocalizedMessage());
|
|
// if exception thrown we can't init the queue, maybe due to name violation. That won't get better, delete it.
|
|
FileUtils.deletedelete(new File(hostsPath, hoststr));
|
|
}
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public synchronized void close() {
|
|
if (depthCache != null) {
|
|
depthCache.clear();
|
|
}
|
|
for (HostQueue queue: this.queues.values()) queue.close();
|
|
this.queues.clear();
|
|
}
|
|
|
|
@Override
|
|
public void clear() {
|
|
if (depthCache != null) {
|
|
depthCache.clear();
|
|
}
|
|
for (HostQueue queue: this.queues.values()) queue.clear();
|
|
this.queues.clear();
|
|
}
|
|
|
|
@Override
|
|
public Request get(final byte[] urlhash) throws IOException {
|
|
String hosthash = ASCII.String(urlhash, 6, 6);
|
|
HostQueue queue = this.queues.get(hosthash);
|
|
if (queue == null) return null;
|
|
return queue.get(urlhash);
|
|
}
|
|
|
|
@Override
|
|
public int removeAllByProfileHandle(final String profileHandle, final long timeout) throws IOException, SpaceExceededException {
|
|
int c = 0;
|
|
for (HostQueue queue: this.queues.values()) {
|
|
c += queue.removeAllByProfileHandle(profileHandle, timeout);
|
|
}
|
|
return c;
|
|
}
|
|
|
|
/**
|
|
* delete all urls which are stored for given host hashes
|
|
* @param hosthashes
|
|
* @return number of deleted urls
|
|
*/
|
|
@Override
|
|
public int removeAllByHostHashes(final Set<String> hosthashes) {
|
|
int c = 0;
|
|
for (String h: hosthashes) {
|
|
HostQueue hq = this.queues.get(h);
|
|
if (hq != null) c += hq.removeAllByHostHashes(hosthashes);
|
|
}
|
|
// remove from cache
|
|
Iterator<Map.Entry<byte[], Long>> i = depthCache.iterator();
|
|
ArrayList<String> deleteHashes = new ArrayList<String>();
|
|
while (i.hasNext()) {
|
|
String h = ASCII.String(i.next().getKey());
|
|
if (hosthashes.contains(h.substring(6))) deleteHashes.add(h);
|
|
}
|
|
for (String h: deleteHashes) depthCache.remove(ASCII.getBytes(h));
|
|
return c;
|
|
}
|
|
|
|
@Override
|
|
public synchronized int remove(final HandleSet urlHashes) throws IOException {
|
|
Map<String, HandleSet> removeLists = new ConcurrentHashMap<String, HandleSet>();
|
|
for (byte[] urlhash: urlHashes) {
|
|
depthCache.remove(urlhash);
|
|
String hosthash = ASCII.String(urlhash, 6, 6);
|
|
HandleSet removeList = removeLists.get(hosthash);
|
|
if (removeList == null) {
|
|
removeList = new RowHandleSet(Word.commonHashLength, Base64Order.enhancedCoder, 100);
|
|
removeLists.put(hosthash, removeList);
|
|
}
|
|
try {removeList.put(urlhash);} catch (SpaceExceededException e) {}
|
|
}
|
|
int c = 0;
|
|
for (Map.Entry<String, HandleSet> entry: removeLists.entrySet()) {
|
|
HostQueue queue = this.queues.get(entry.getKey());
|
|
if (queue != null) c += queue.remove(entry.getValue());
|
|
}
|
|
return c;
|
|
}
|
|
|
|
/**
|
|
* @return true when the URL is queued is this or any other HostBalancer
|
|
* instance (as {@link #depthCache} is shared between all HostBalancer
|
|
* instances)
|
|
*/
|
|
@Override
|
|
public boolean has(final byte[] urlhashb) {
|
|
if (depthCache.has(urlhashb)) return true;
|
|
String hosthash = ASCII.String(urlhashb, 6, 6);
|
|
HostQueue queue = this.queues.get(hosthash);
|
|
if (queue == null) return false;
|
|
return queue.has(urlhashb);
|
|
}
|
|
|
|
@Override
|
|
public int size() {
|
|
int c = 0;
|
|
for (HostQueue queue: this.queues.values()) {
|
|
c += queue.size();
|
|
}
|
|
return c;
|
|
}
|
|
|
|
@Override
|
|
public boolean isEmpty() {
|
|
for (HostQueue queue: this.queues.values()) {
|
|
if (!queue.isEmpty()) return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
@Override
|
|
public int getOnDemandLimit() {
|
|
return this.onDemandLimit;
|
|
}
|
|
|
|
@Override
|
|
public boolean getExceed134217727() {
|
|
return this.exceed134217727;
|
|
}
|
|
/**
|
|
* push a request to one of the host queues. If the queue does not exist, it is created
|
|
* @param entry
|
|
* @param profile
|
|
* @param robots
|
|
* @return null if everything is ok or a string with an error message if the push is not allowed according to the crawl profile or robots
|
|
* @throws IOException
|
|
* @throws SpaceExceededException
|
|
*/
|
|
@Override
|
|
public String push(final Request entry, CrawlProfile profile, final RobotsTxt robots) throws IOException, SpaceExceededException {
|
|
if (this.has(entry.url().hash())) return "double occurrence";
|
|
depthCache.put(entry.url().hash(), entry.depth());
|
|
String hosthash = entry.url().hosthash();
|
|
synchronized (this) {
|
|
HostQueue queue = this.queues.get(hosthash);
|
|
if (queue == null) {
|
|
queue = new HostQueue(this.hostsPath, entry.url(), this.queues.size() > this.onDemandLimit, this.exceed134217727);
|
|
this.queues.put(hosthash, queue);
|
|
// profile might be null when continue crawls after YaCy restart
|
|
robots.ensureExist(entry.url(), profile == null ? ClientIdentification.yacyInternetCrawlerAgent : profile.getAgent(), true); // concurrently load all robots.txt
|
|
}
|
|
return queue.push(entry, profile, robots);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* get the next entry in this crawl queue in such a way that the domain access time delta is maximized
|
|
* and always above the given minimum delay time. An additional delay time is computed using the robots.txt
|
|
* crawl-delay time which is always respected. In case the minimum time cannot ensured, this method pauses
|
|
* the necessary time until the url is released and returned as CrawlEntry object. In case that a profile
|
|
* for the computed Entry does not exist, null is returned
|
|
* @param delay true if the requester demands forced delays using explicit thread sleep
|
|
* @param profile
|
|
* @return a url in a CrawlEntry object
|
|
* @throws IOException
|
|
* @throws SpaceExceededException
|
|
*/
|
|
@Override
|
|
public Request pop(boolean delay, CrawlSwitchboard cs, RobotsTxt robots) throws IOException {
|
|
tryagain: while (true) try {
|
|
HostQueue rhq = null;
|
|
String rhh = null;
|
|
|
|
synchronized (this) {
|
|
if (this.roundRobinHostHashes.size() == 0) {
|
|
// refresh the round-robin cache
|
|
this.roundRobinHostHashes.addAll(this.queues.keySet());
|
|
// quickly get rid of small stacks to reduce number of files:
|
|
// remove all stacks with more than 10 entries
|
|
// this shall kick out small stacks to prevent that too many files are opened for very wide crawls
|
|
boolean smallStacksExist = false;
|
|
boolean singletonStacksExist = false;
|
|
smallsearch: for (String s: this.roundRobinHostHashes) {
|
|
HostQueue hq = this.queues.get(s);
|
|
if (hq != null) {
|
|
int size = hq.size();
|
|
if (size == 1) {singletonStacksExist = true; break smallsearch;}
|
|
if (size <= 10) {smallStacksExist = true; break smallsearch;}
|
|
}
|
|
}
|
|
if (singletonStacksExist || smallStacksExist) {
|
|
Iterator<String> i = this.roundRobinHostHashes.iterator();
|
|
smallstacks: while (i.hasNext()) {
|
|
if (this.roundRobinHostHashes.size() <= 10) break smallstacks; // don't shrink the hosts until nothing is left
|
|
String s = i.next();
|
|
HostQueue hq = this.queues.get(s);
|
|
if (hq == null) {i.remove(); continue smallstacks;}
|
|
int delta = Latency.waitingRemainingGuessed(hq.getHost(), hq.getPort(), s, robots, ClientIdentification.yacyInternetCrawlerAgent);
|
|
if (delta < 0) continue; // keep all non-waiting stacks; they are useful to speed up things
|
|
// to protect all small stacks which have a fast throughput, remove all with long waiting time
|
|
if (delta >= 1000) {i.remove(); continue smallstacks;}
|
|
int size = hq.size();
|
|
if (singletonStacksExist) {
|
|
if (size != 1) {i.remove(); continue smallstacks;} // remove all non-singletons
|
|
} else /*smallStacksExist*/ {
|
|
if (size > 10) {i.remove(); continue smallstacks;} // remove all large stacks
|
|
}
|
|
}
|
|
}
|
|
if (this.roundRobinHostHashes.size() == 1) {
|
|
if (log.isFine()) log.fine("(re-)initialized the round-robin queue with one host");
|
|
} else {
|
|
log.info("(re-)initialized the round-robin queue; " + this.roundRobinHostHashes.size() + " hosts.");
|
|
}
|
|
}
|
|
if (this.roundRobinHostHashes.size() == 0) return null;
|
|
|
|
// if the queue size is 1, just take that
|
|
if (this.roundRobinHostHashes.size() == 1) {
|
|
rhh = this.roundRobinHostHashes.iterator().next();
|
|
rhq = this.queues.get(rhh);
|
|
}
|
|
|
|
if (rhq == null) {
|
|
// mixed minimum sleep time / largest queue strategy:
|
|
// create a map of sleep time / queue relations with a fuzzy sleep time (ms / 500).
|
|
// if the entry with the smallest sleep time contains at least two entries,
|
|
// then the larger one from these queues are selected.
|
|
TreeMap<Integer, List<String>> fastTree = new TreeMap<>();
|
|
mixedstrategy: for (String h: this.roundRobinHostHashes) {
|
|
HostQueue hq = this.queues.get(h);
|
|
if (hq != null) {
|
|
int delta = Latency.waitingRemainingGuessed(hq.getHost(), hq.getPort(), h, robots, ClientIdentification.yacyInternetCrawlerAgent) / 200;
|
|
if (delta < 0) delta = 0;
|
|
List<String> queueHashes = fastTree.get(delta);
|
|
if (queueHashes == null) {
|
|
queueHashes = new ArrayList<>(2);
|
|
fastTree.put(delta, queueHashes);
|
|
}
|
|
queueHashes.add(h);
|
|
// check stop criteria
|
|
List<String> firstEntries = fastTree.firstEntry().getValue();
|
|
if (firstEntries.size() > 1) {
|
|
// select larger queue from that list
|
|
int largest = Integer.MIN_VALUE;
|
|
for (String hh: firstEntries) {
|
|
HostQueue hhq = this.queues.get(hh);
|
|
if (hhq != null) {
|
|
int s = hhq.size();
|
|
if (s > largest) {
|
|
largest = s;
|
|
rhh = hh;
|
|
}
|
|
}
|
|
}
|
|
rhq = this.queues.get(rhh);
|
|
break mixedstrategy;
|
|
}
|
|
}
|
|
}
|
|
if (rhq == null && fastTree.size() > 0) {
|
|
// it may be possible that the lowest entry never has more than one queues assigned
|
|
// in this case just take the smallest entry
|
|
List<String> firstEntries = fastTree.firstEntry().getValue();
|
|
assert firstEntries.size() == 1;
|
|
rhh = firstEntries.get(0);
|
|
rhq = this.queues.get(rhh);
|
|
}
|
|
// to prevent that the complete roundrobinhosthashes are taken for each round, we remove the entries from the top of the fast queue
|
|
List<String> lastEntries = fastTree.size() > 0 ? fastTree.lastEntry().getValue() : null;
|
|
if (lastEntries != null) {
|
|
for (String h: lastEntries) this.roundRobinHostHashes.remove(h);
|
|
}
|
|
}
|
|
|
|
/*
|
|
// first strategy: get one entry which does not need sleep time
|
|
Iterator<String> nhhi = this.roundRobinHostHashes.iterator();
|
|
nosleep: while (nhhi.hasNext()) {
|
|
rhh = nhhi.next();
|
|
rhq = this.queues.get(rhh);
|
|
if (rhq == null) {
|
|
nhhi.remove();
|
|
continue nosleep;
|
|
}
|
|
int delta = Latency.waitingRemainingGuessed(rhq.getHost(), rhh, robots, ClientIdentification.yacyInternetCrawlerAgent);
|
|
if (delta <= 10 || this.roundRobinHostHashes.size() == 1 || rhq.size() == 1) {
|
|
nhhi.remove();
|
|
break nosleep;
|
|
}
|
|
}
|
|
if (rhq == null) {
|
|
// second strategy: take from the largest stack
|
|
int largest = Integer.MIN_VALUE;
|
|
for (String h: this.roundRobinHostHashes) {
|
|
HostQueue hq = this.queues.get(h);
|
|
if (hq != null) {
|
|
int s = hq.size();
|
|
if (s > largest) {
|
|
largest = s;
|
|
rhh = h;
|
|
}
|
|
}
|
|
}
|
|
rhq = this.queues.get(rhh);
|
|
}
|
|
*/
|
|
}
|
|
|
|
if (rhq == null) {
|
|
this.roundRobinHostHashes.clear(); // force re-initialization
|
|
continue tryagain;
|
|
}
|
|
this.roundRobinHostHashes.remove(rhh); // prevent that the queue is used again
|
|
long timestamp = System.currentTimeMillis();
|
|
Request request = rhq.pop(delay, cs, robots); // this pop is outside of synchronization to prevent blocking of pushes
|
|
long actualwaiting = System.currentTimeMillis() - timestamp;
|
|
|
|
if (actualwaiting > 1000) {
|
|
synchronized (this) {
|
|
// to prevent that this occurs again, remove all stacks with positive delay times (which may be less after that waiting)
|
|
Iterator<String> i = this.roundRobinHostHashes.iterator();
|
|
protectcheck: while (i.hasNext()) {
|
|
if (this.roundRobinHostHashes.size() <= 3) break protectcheck; // don't shrink the hosts until nothing is left
|
|
String s = i.next();
|
|
HostQueue hq = this.queues.get(s);
|
|
if (hq == null) {i.remove(); continue protectcheck;}
|
|
int delta = Latency.waitingRemainingGuessed(hq.getHost(), hq.getPort(), s, robots, ClientIdentification.yacyInternetCrawlerAgent);
|
|
if (delta >= 0) {i.remove();}
|
|
}
|
|
}
|
|
}
|
|
|
|
if (rhq.isEmpty()) {
|
|
synchronized (this) {
|
|
this.queues.remove(rhh);
|
|
}
|
|
rhq.close();
|
|
}
|
|
if (request == null) continue tryagain;
|
|
return request;
|
|
} catch (ConcurrentModificationException e) {
|
|
continue tryagain;
|
|
} catch (IOException e) {
|
|
throw e;
|
|
} catch (Throwable e) {
|
|
ConcurrentLog.logException(e);
|
|
throw new IOException(e.getMessage());
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public Iterator<Request> iterator() throws IOException {
|
|
final Iterator<HostQueue> hostsIterator = this.queues.values().iterator();
|
|
@SuppressWarnings("unchecked")
|
|
final Iterator<Request>[] hostIterator = (Iterator<Request>[]) Array.newInstance(Iterator.class, 1);
|
|
hostIterator[0] = null;
|
|
return new Iterator<Request>() {
|
|
@Override
|
|
public boolean hasNext() {
|
|
return hostsIterator.hasNext() || (hostIterator[0] != null && hostIterator[0].hasNext());
|
|
}
|
|
@Override
|
|
public Request next() {
|
|
synchronized (HostBalancer.this) {
|
|
while (hostIterator[0] == null || !hostIterator[0].hasNext()) try {
|
|
HostQueue entry = hostsIterator.next();
|
|
hostIterator[0] = entry.iterator();
|
|
} catch (IOException e) {}
|
|
if (!hostIterator[0].hasNext()) return null;
|
|
return hostIterator[0].next();
|
|
}
|
|
}
|
|
@Override
|
|
public void remove() {
|
|
hostIterator[0].remove();
|
|
}
|
|
};
|
|
}
|
|
|
|
/**
|
|
* get a list of domains that are currently maintained as domain stacks
|
|
* @return a map of clear text strings of host names + ports to an integer array: {the size of the domain stack, guessed delta waiting time}
|
|
*/
|
|
@Override
|
|
public Map<String, Integer[]> getDomainStackHosts(RobotsTxt robots) {
|
|
Map<String, Integer[]> map = new TreeMap<String, Integer[]>(); // we use a tree map to get a stable ordering
|
|
for (HostQueue hq: this.queues.values()) {
|
|
int delta = Latency.waitingRemainingGuessed(hq.getHost(), hq.getPort(), hq.getHostHash(), robots, ClientIdentification.yacyInternetCrawlerAgent);
|
|
map.put(hq.getHost() + ":" + hq.getPort(), new Integer[]{hq.size(), delta});
|
|
}
|
|
return map;
|
|
}
|
|
|
|
/**
|
|
* get lists of crawl request entries for a specific host
|
|
* @param host
|
|
* @param maxcount
|
|
* @param maxtime
|
|
* @return a list of crawl loader requests
|
|
*/
|
|
@Override
|
|
public List<Request> getDomainStackReferences(String host, int maxcount, long maxtime) {
|
|
if (host == null) {
|
|
return Collections.emptyList();
|
|
}
|
|
try {
|
|
HostQueue hq = this.queues.get(DigestURL.hosthash(host, host.startsWith("ftp.") ? 21 : 80));
|
|
if (hq == null) hq = this.queues.get(DigestURL.hosthash(host, 443));
|
|
return hq == null ? new ArrayList<Request>(0) : hq.getDomainStackReferences(host, maxcount, maxtime);
|
|
} catch (final MalformedURLException e) {
|
|
ConcurrentLog.logException(e);
|
|
return Collections.emptyList();
|
|
}
|
|
}
|
|
|
|
}
|