added a fast-fail concept in search processes. The search now has better control if all the remote searches may bring any result. If all processes are finished, then all search tasks fail fast.

git-svn-id: https://svn.berlios.de/svnroot/repos/yacy/trunk@6290 6c8d7289-2bf4-0310-a012-ef5d649a1542

htroot/IndexControlRWIs_p.java

@@ -366,7 +366,7 @@ public class IndexControlRWIs_p {
             URLMetadataRow entry;
             String us;
             long rn = -1;
-            while ((ranked.size() > 0) && ((entry = ranked.takeURL(false)) != null)) {
+            while ((ranked.size() > 0) && ((entry = ranked.takeURL(false, 60000)) != null)) {
                 if ((entry == null) || (entry.metadata() == null)) continue;
                 url = entry.metadata().url();
                 if (url == null) continue;

source/de/anomic/kelondro/text/metadataPrototype/URLMetadataRow.java

@@ -123,6 +123,15 @@ public class URLMetadataRow implements Metadata {
     private final long ranking; // during generation of a search result this value is set
     private Components comp;
     
+    public URLMetadataRow() {
+        // create a dummy entry, good to produce poison objects
+        this.entry = rowdef.newEntry();
+        this.snippet = null;
+        this.word = null;
+        this.ranking = 0;
+        this.comp = null;
+    }
+    
     public URLMetadataRow(
             final yacyURL url,
             final String dc_title,
@@ -145,7 +154,7 @@ public class URLMetadataRow implements Metadata {
             final int limage,
             final int lvideo,
             final int lapp) {
-        // create new entry and store it into database
+        // create new entry
         this.entry = rowdef.newEntry();
         this.entry.setCol(col_hash, url.hash(), null);
         this.entry.setCol(col_comp, encodeComp(url, dc_title, dc_creator, dc_subject, ETag));

source/de/anomic/search/RankingProcess.java

@@ -78,6 +78,7 @@ public final class RankingProcess extends Thread {
     
     private int remote_peerCount, remote_indexCount, remote_resourceSize, local_resourceSize;
     private final SortStack<WordReferenceVars> stack;
+    private int feeders;
     private final HashMap<String, SortStack<WordReferenceVars>> doubleDomCache; // key = domhash (6 bytes); value = like stack
     private final HashSet<String> handover; // key = urlhash; used for double-check of urls that had been handed over to search process
     
@@ -85,6 +86,7 @@ public final class RankingProcess extends Thread {
     private final ConcurrentHashMap<String, HostInfo> hostNavigator;
     private final ConcurrentHashMap<String, AuthorInfo> authorNavigator;
     
+    
     public RankingProcess(
             final Segment indexSegment,
             final QueryParams query,
@@ -114,6 +116,8 @@ public final class RankingProcess extends Thread {
         this.ref = new ConcurrentHashMap<String, Integer>();
         this.domZones = new int[8];
         for (int i = 0; i < 8; i++) {this.domZones[i] = 0;}
+        this.feeders = concurrency;
+        assert this.feeders >= 1;
     }
     
     public void run() {
@@ -140,6 +144,7 @@ public final class RankingProcess extends Thread {
         } catch (final Exception e) {
             e.printStackTrace();
         }
+        oneFeederTerminated();
     }
     
     public long ranking(final WordReferenceVars word) {
@@ -262,6 +267,22 @@ public final class RankingProcess extends Thread {
         serverProfiling.update("SEARCH", new ProfilingGraph.searchEvent(query.id(true), SearchEvent.PRESORT, index.size(), System.currentTimeMillis() - timer), false);
     }
     
+    /**
+     * method to signal the incoming stack that one feeder has terminated
+     */
+    public void oneFeederTerminated() {
+    	this.feeders--;
+    	assert this.feeders >= 0 : "feeders = " + this.feeders;
+    }
+    
+    public void moreFeeders(final int countMoreFeeders) {
+    	this.feeders += countMoreFeeders;
+    }
+    
+    public boolean feedingIsFinished() {
+    	return this.feeders == 0;
+    }
+    
     private boolean testFlags(final WordReference ientry) {
         if (query.constraint == null) return true;
         // test if ientry matches with filter
@@ -337,12 +358,16 @@ public final class RankingProcess extends Thread {
         return bestEntry;
     }
     
-    public URLMetadataRow takeURL(final boolean skipDoubleDom) {
+    public URLMetadataRow takeURL(final boolean skipDoubleDom, final long timeout) {
         // returns from the current RWI list the best URL entry and removes this entry from the list
-        while ((stack.size() > 0) || (size() > 0)) {
-            if (((stack.size() == 0) && (size() == 0))) break;
+    	long timeLimit = System.currentTimeMillis() + timeout;
+    	while (System.currentTimeMillis() < timeLimit) {
             final SortStack<WordReferenceVars>.stackElement obrwi = takeRWI(skipDoubleDom);
-            if (obrwi == null) continue; // *** ? this happened and the thread was suspended silently. cause?
+            if (obrwi == null) {
+            	if (this.feedingIsFinished()) return null;
+            	try {Thread.sleep(50);} catch (final InterruptedException e1) {}
+            	continue;
+            }
             final URLMetadataRow page = indexSegment.urlMetadata().load(obrwi.element.metadataHash(), obrwi.element, obrwi.weight.longValue());
             if (page == null) {
             	misses.add(obrwi.element.metadataHash());
@@ -421,18 +446,6 @@ public final class RankingProcess extends Thread {
         return null;
     }
     
-    public URLMetadataRow takeURL(final boolean skipDoubleDom, long timeout) {
-        timeout += System.currentTimeMillis();
-        long wait = 10;
-        while (System.currentTimeMillis() < timeout) {
-            URLMetadataRow row = takeURL(skipDoubleDom);
-            if (row != null) return row;
-            try {Thread.sleep(wait);} catch (final InterruptedException e1) {}
-            wait = wait * 2;
-        }
-        return null;
-    }
-    
     public int size() {
         //assert sortedRWIEntries.size() == urlhashes.size() : "sortedRWIEntries.size() = " + sortedRWIEntries.size() + ", urlhashes.size() = " + urlhashes.size();
         int c = stack.size();

source/de/anomic/search/SearchEvent.java

@@ -104,17 +104,17 @@ public final class SearchEvent {
         final long start = System.currentTimeMillis();
         if ((query.domType == QueryParams.SEARCHDOM_GLOBALDHT) ||
             (query.domType == QueryParams.SEARCHDOM_CLUSTERALL)) {
+        	final int fetchpeers = 12;
             
         	// initialize a ranking process that is the target for data
         	// that is generated concurrently from local and global search threads
-            this.rankedCache = new RankingProcess(indexSegment, query, max_results_preparation, 16);
+            this.rankedCache = new RankingProcess(indexSegment, query, max_results_preparation, fetchpeers + 1);
             
             // start a local search concurrently
             this.rankedCache.start();
                        
             // start global searches
             final long timer = System.currentTimeMillis();
-            final int fetchpeers = 12;
             Log.logFine("SEARCH_EVENT", "STARTING " + fetchpeers + " THREADS TO CATCH EACH " + query.displayResults() + " URLs");
             this.primarySearchThreads = (query.queryHashes.size() == 0) ? null : yacySearch.primaryRemoteSearches(
                     QueryParams.hashSet2hashString(query.queryHashes),
@@ -137,6 +137,7 @@ public final class SearchEvent {
                     query.ranking,
                     query.constraint,
                     (query.domType == QueryParams.SEARCHDOM_GLOBALDHT) ? null : preselectedPeerHashes);
+            if (this.primarySearchThreads.length > fetchpeers) this.rankedCache.moreFeeders(this.primarySearchThreads.length - fetchpeers);
             serverProfiling.update("SEARCH", new ProfilingGraph.searchEvent(query.id(true), "remote search thread start", this.primarySearchThreads.length, System.currentTimeMillis() - timer), false);
             
             // finished searching

source/de/anomic/yacy/yacySearch.java

@@ -126,6 +126,8 @@ public class yacySearch extends Thread {
             }
         } catch (final Exception e) {
             e.printStackTrace();
+        } finally {
+        	containerCache.oneFeederTerminated();
         }
     }