boxtec
/
yacy_search_server
mirror of https://github.com/yacy/yacy_search_server
4
1
Fork 0
Code Issues Projects Releases Wiki Activity

enhanced strategy in host browser

Browse Source 
limit number of fresh hosts in round robin hashes
pull/402/head
Michael Peter Christen 4 years ago
parent 9be36800a4
commit 63f58e4785
1 changed files with 40 additions and 29 deletions
Show Stats Download Patch File Download Diff File

69
source/net/yacy/crawler/HostBalancer.java
Unescape View File

@ -65,7 +65,7 @@ public class HostBalancer implements Balancer {
private final static ConcurrentLog log = new ConcurrentLog("HostBalancer"); 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"); public final static HandleMap depthCache = new RowHandleMap(Word.commonHashLength, Word.commonHashOrder, 2, 8 * 1024 * 1024, "HostBalancer.DepthCache");
private final File hostsPath; private final File hostsPath;
private final boolean exceed134217727; private final boolean exceed134217727;
private final Map<String, HostQueue> queues; private final Map<String, HostQueue> queues;
@ -84,7 +84,7 @@ public class HostBalancer implements Balancer {
final boolean exceed134217727) { final boolean exceed134217727) {
this(hostsPath, onDemandLimit, exceed134217727, true); this(hostsPath, onDemandLimit, exceed134217727, true);
} }
/** /**
* Create a new instance and fills the queue by scanning the hostsPath directory. * Create a new instance and fills the queue by scanning the hostsPath directory.
* @param hostsPath * @param hostsPath
@ -100,7 +100,7 @@ public class HostBalancer implements Balancer {
this.hostsPath = hostsPath; this.hostsPath = hostsPath;
this.onDemandLimit = onDemandLimit; this.onDemandLimit = onDemandLimit;
this.exceed134217727 = exceed134217727; this.exceed134217727 = exceed134217727;
// create a stack for newly entered entries // create a stack for newly entered entries
if (!(hostsPath.exists())) hostsPath.mkdirs(); // make the path if (!(hostsPath.exists())) hostsPath.mkdirs(); // make the path
this.queues = new ConcurrentHashMap<String, HostQueue>(); this.queues = new ConcurrentHashMap<String, HostQueue>();
@ -114,7 +114,7 @@ public class HostBalancer implements Balancer {
* return immediately (as large unfinished crawls may take longer to load) * return immediately (as large unfinished crawls may take longer to load)
*/ */
private void init(final boolean async) { private void init(final boolean async) {
if(async) { if(async) {
Thread t = new Thread("HostBalancer.init") { Thread t = new Thread("HostBalancer.init") {
@Override @Override
public void run() { public void run() {
@ -122,10 +122,10 @@ public class HostBalancer implements Balancer {
} }
}; };
t.start(); t.start();
} else { } else {
runInit(); runInit();
} }
} }
/** /**
@ -185,7 +185,7 @@ public class HostBalancer implements Balancer {
} }
return c; return c;
} }
/** /**
* delete all urls which are stored for given host hashes * delete all urls which are stored for given host hashes
* @param hosthashes * @param hosthashes
@ -230,11 +230,11 @@ public class HostBalancer implements Balancer {
return c; return c;
} }
/** /**
* @return true when the URL is queued is this or any other HostBalancer * @return true when the URL is queued is this or any other HostBalancer
* instance (as {@link #depthCache} is shared between all HostBalancer * instance (as {@link #depthCache} is shared between all HostBalancer
* instances) * instances)
*/ */
@Override @Override
public boolean has(final byte[] urlhashb) { public boolean has(final byte[] urlhashb) {
if (depthCache.has(urlhashb)) return true; if (depthCache.has(urlhashb)) return true;
@ -313,7 +313,7 @@ public class HostBalancer implements Balancer {
tryagain: while (true) try { tryagain: while (true) try {
HostQueue rhq = null; HostQueue rhq = null;
String rhh = null; String rhh = null;
synchronized (this) { synchronized (this) {
if (this.roundRobinHostHashes.size() == 0) { if (this.roundRobinHostHashes.size() == 0) {
// refresh the round-robin cache // refresh the round-robin cache
@ -331,14 +331,21 @@ public class HostBalancer implements Balancer {
if (size <= 10) {smallStacksExist = true; break smallsearch;} if (size <= 10) {smallStacksExist = true; break smallsearch;}
} }
} }
if (singletonStacksExist || smallStacksExist) { Set<String> freshhosts = new HashSet<>();
Iterator<String> i = this.roundRobinHostHashes.iterator(); Iterator<String> i = this.roundRobinHostHashes.iterator();
smallstacks: while (i.hasNext()) { smallstacks: while (i.hasNext()) {
if (this.roundRobinHostHashes.size() <= 10) break smallstacks; // don't shrink the hosts until nothing is left if (this.roundRobinHostHashes.size() <= 10) break smallstacks; // don't shrink the hosts until nothing is left
String s = i.next(); String hosthash = i.next();
HostQueue hq = this.queues.get(s); HostQueue hq = this.queues.get(hosthash);
if (hq == null) {i.remove(); continue smallstacks;} if (hq == null) {i.remove(); continue smallstacks;}
int delta = Latency.waitingRemainingGuessed(hq.getHost(), hq.getPort(), s, robots, ClientIdentification.yacyInternetCrawlerAgent); int delta = Latency.waitingRemainingGuessed(hq.getHost(), hq.getPort(), hosthash, robots, ClientIdentification.yacyInternetCrawlerAgent);
if (delta == Integer.MIN_VALUE) {
// never-crawled hosts; we do not want to have too many of them in here. Loading new hosts means: waiting for robots.txt to load
freshhosts.add(hosthash);
i.remove();
continue smallstacks;
}
if (singletonStacksExist || smallStacksExist) {
if (delta < 0) continue; // keep all non-waiting stacks; they are useful to speed up things 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 // to protect all small stacks which have a fast throughput, remove all with long waiting time
if (delta >= 1000) {i.remove(); continue smallstacks;} if (delta >= 1000) {i.remove(); continue smallstacks;}
@ -350,6 +357,10 @@ public class HostBalancer implements Balancer {
} }
} }
} }
// put at least one of the fresh hosts back
if (freshhosts.size() > 0) this.roundRobinHostHashes.add(freshhosts.iterator().next());
// result
if (this.roundRobinHostHashes.size() == 1) { if (this.roundRobinHostHashes.size() == 1) {
if (log.isFine()) log.fine("(re-)initialized the round-robin queue with one host"); if (log.isFine()) log.fine("(re-)initialized the round-robin queue with one host");
} else { } else {
@ -357,13 +368,13 @@ public class HostBalancer implements Balancer {
} }
} }
if (this.roundRobinHostHashes.size() == 0) return null; if (this.roundRobinHostHashes.size() == 0) return null;
// if the queue size is 1, just take that // if the queue size is 1, just take that
if (this.roundRobinHostHashes.size() == 1) { if (this.roundRobinHostHashes.size() == 1) {
rhh = this.roundRobinHostHashes.iterator().next(); rhh = this.roundRobinHostHashes.iterator().next();
rhq = this.queues.get(rhh); rhq = this.queues.get(rhh);
} }
if (rhq == null) { if (rhq == null) {
// mixed minimum sleep time / largest queue strategy: // mixed minimum sleep time / largest queue strategy:
// create a map of sleep time / queue relations with a fuzzy sleep time (ms / 500). // create a map of sleep time / queue relations with a fuzzy sleep time (ms / 500).
@ -449,7 +460,7 @@ public class HostBalancer implements Balancer {
} }
*/ */
} }
if (rhq == null) { if (rhq == null) {
this.roundRobinHostHashes.clear(); // force re-initialization this.roundRobinHostHashes.clear(); // force re-initialization
continue tryagain; continue tryagain;
@ -458,7 +469,7 @@ public class HostBalancer implements Balancer {
long timestamp = System.currentTimeMillis(); long timestamp = System.currentTimeMillis();
Request request = rhq.pop(delay, cs, robots); // this pop is outside of synchronization to prevent blocking of pushes Request request = rhq.pop(delay, cs, robots); // this pop is outside of synchronization to prevent blocking of pushes
long actualwaiting = System.currentTimeMillis() - timestamp; long actualwaiting = System.currentTimeMillis() - timestamp;
if (actualwaiting > 1000) { if (actualwaiting > 1000) {
synchronized (this) { synchronized (this) {
// to prevent that this occurs again, remove all stacks with positive delay times (which may be less after that waiting) // to prevent that this occurs again, remove all stacks with positive delay times (which may be less after that waiting)
@ -473,7 +484,7 @@ public class HostBalancer implements Balancer {
} }
} }
} }
if (rhq.isEmpty()) { if (rhq.isEmpty()) {
synchronized (this) { synchronized (this) {
this.queues.remove(rhh); this.queues.remove(rhh);
@ -545,7 +556,7 @@ public class HostBalancer implements Balancer {
@Override @Override
public List<Request> getDomainStackReferences(String host, int maxcount, long maxtime) { public List<Request> getDomainStackReferences(String host, int maxcount, long maxtime) {
if (host == null) { if (host == null) {
return Collections.emptyList(); return Collections.emptyList();
} }
try { try {
HostQueue hq = this.queues.get(DigestURL.hosthash(host, host.startsWith("ftp.") ? 21 : 80)); HostQueue hq = this.queues.get(DigestURL.hosthash(host, host.startsWith("ftp.") ? 21 : 80));

Write Preview
Loading…
Cancel
Save