2005-10-10 02:32:15 +02:00
// plasmaSearchEvent.java
// -----------------------
// part of YACY
// (C) by Michael Peter Christen; mc@anomic.de
// first published on http://www.anomic.de
// Frankfurt, Germany, 2005
// Created: 10.10.2005
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
//
// This program 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 General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
//
// Using this software in any meaning (reading, learning, copying, compiling,
// running) means that you agree that the Author(s) is (are) not responsible
// for cost, loss of data or any harm that may be caused directly or indirectly
// by usage of this softare or this documentation. The usage of this software
// is on your own risk. The installation and usage (starting/running) of this
// software may allow other people or application to access your computer and
// any attached devices and is highly dependent on the configuration of the
// software which must be done by the user of the software; the author(s) is
// (are) also not responsible for proper configuration and usage of the
// software, even if provoked by documentation provided together with
// the software.
//
// Any changes to this file according to the GPL as documented in the file
// gpl.txt aside this file in the shipment you received can be done to the
// lines that follows this copyright notice here, but changes must not be
// done inside the copyright notive above. A re-distribution must contain
// the intact and unchanged copyright notice.
// Contributions and changes to the program code must be marked as such.
package de.anomic.plasma ;
2005-10-11 09:06:33 +02:00
import java.util.Iterator ;
2005-10-12 14:28:49 +02:00
import java.util.Set ;
import java.util.HashSet ;
import java.util.TreeMap ;
import java.util.Enumeration ;
import java.io.IOException ;
import de.anomic.kelondro.kelondroException ;
import de.anomic.server.logging.serverLog ;
import de.anomic.server.serverCodings ;
2005-10-13 15:57:15 +02:00
import de.anomic.server.serverInstantThread ;
import de.anomic.yacy.yacySearch ;
2005-10-10 02:32:15 +02:00
public final class plasmaSearchEvent {
2005-10-24 02:34:15 +02:00
public static plasmaSearchEvent lastEvent = null ;
2005-10-12 14:28:49 +02:00
private serverLog log ;
2005-10-11 09:06:33 +02:00
private plasmaSearchQuery query ;
2005-10-12 14:28:49 +02:00
private plasmaWordIndex wordIndex ;
private plasmaCrawlLURL urlStore ;
private plasmaSnippetCache snippetCache ;
2005-10-13 15:57:15 +02:00
private plasmaWordIndexEntity rcLocal , rcGlobal ; // caches for results
2005-10-23 19:50:27 +02:00
private plasmaSearchProfile profileLocal , profileGlobal ;
2005-10-13 15:57:15 +02:00
private yacySearch [ ] searchThreads ;
2005-10-12 14:28:49 +02:00
public plasmaSearchEvent ( plasmaSearchQuery query , serverLog log , plasmaWordIndex wordIndex , plasmaCrawlLURL urlStore , plasmaSnippetCache snippetCache ) {
this . log = log ;
this . wordIndex = wordIndex ;
2005-10-11 09:06:33 +02:00
this . query = query ;
2005-10-12 14:28:49 +02:00
this . urlStore = urlStore ;
this . snippetCache = snippetCache ;
2005-10-13 15:57:15 +02:00
this . rcLocal = new plasmaWordIndexEntity ( null ) ;
this . rcGlobal = new plasmaWordIndexEntity ( null ) ;
2005-10-23 19:50:27 +02:00
if ( query . domType = = plasmaSearchQuery . SEARCHDOM_GLOBALDHT ) {
this . profileLocal = new plasmaSearchProfile ( 4 * query . maximumTime / 10 , query . wantedResults ) ;
this . profileGlobal = new plasmaSearchProfile ( 6 * query . maximumTime / 10 , query . wantedResults ) ;
} else {
this . profileLocal = new plasmaSearchProfile ( query . maximumTime , query . wantedResults ) ;
this . profileGlobal = null ;
}
2005-10-13 15:57:15 +02:00
this . searchThreads = null ;
2005-10-12 14:28:49 +02:00
}
2005-10-24 02:34:15 +02:00
public plasmaSearchQuery getQuery ( ) {
return query ;
}
public yacySearch [ ] getSearchThreads ( ) {
return searchThreads ;
}
2005-10-13 15:57:15 +02:00
public plasmaSearchResult search ( ) {
// combine all threads
if ( query . domType = = plasmaSearchQuery . SEARCHDOM_GLOBALDHT ) {
2005-10-23 19:50:27 +02:00
int fetchpeers = ( int ) ( query . maximumTime / 1000L ) ; // number of target peers; means 10 peers in 10 seconds
if ( fetchpeers > 10 ) fetchpeers = 10 ;
2005-10-13 15:57:15 +02:00
// remember time
long start = System . currentTimeMillis ( ) ;
// first trigger a local search within a separate thread
serverInstantThread . oneTimeJob ( this , " localSearch " , log , 0 ) ;
// do a global search
2005-10-23 19:50:27 +02:00
int globalContributions = globalSearch ( fetchpeers ) ;
2005-10-13 15:57:15 +02:00
log . logFine ( " SEARCH TIME AFTER GLOBAL-TRIGGER TO " + fetchpeers + " PEERS: " + ( ( System . currentTimeMillis ( ) - start ) / 1000 ) + " seconds " ) ;
try {
// combine the result and order
2005-10-23 19:50:27 +02:00
plasmaSearchResult result = order ( ) ;
2005-10-13 15:57:15 +02:00
result . globalContributions = globalContributions ;
result . localContributions = rcLocal . size ( ) ;
// flush results in a separate thread
serverInstantThread . oneTimeJob ( this , " flushResults " , log , 0 ) ;
// clean up
if ( ( rcLocal ! = null ) & & ( ! ( rcLocal . isTMPEntity ( ) ) ) ) rcLocal . close ( ) ;
rcLocal = null ;
// return search result
2005-10-23 19:50:27 +02:00
log . logFine ( " SEARCHRESULT: " + profileLocal . reportToString ( ) ) ;
2005-10-24 02:34:15 +02:00
lastEvent = this ;
2005-10-13 15:57:15 +02:00
return result ;
} catch ( IOException e ) {
return null ;
}
} else {
// do a local search
long start = System . currentTimeMillis ( ) ;
try {
2005-10-23 19:50:27 +02:00
localSearch ( ) ;
plasmaSearchResult result = order ( ) ;
2005-10-13 15:57:15 +02:00
result . localContributions = rcLocal . size ( ) ;
// clean up
if ( ( rcLocal ! = null ) & & ( ! ( rcLocal . isTMPEntity ( ) ) ) ) rcLocal . close ( ) ;
rcLocal = null ;
2005-10-23 19:50:27 +02:00
// return search result
log . logFine ( " SEARCHRESULT: " + profileLocal . reportToString ( ) ) ;
2005-10-24 02:34:15 +02:00
lastEvent = this ;
2005-10-13 15:57:15 +02:00
return result ;
} catch ( IOException e ) {
return null ;
}
}
}
2005-10-23 19:50:27 +02:00
public int localSearch ( ) throws IOException {
2005-10-12 14:28:49 +02:00
// search for the set of hashes and return an array of urlEntry elements
// retrieve entities that belong to the hashes
2005-10-23 19:50:27 +02:00
profileLocal . startTimer ( ) ;
Set entities = wordIndex . getEntities ( query . queryHashes , true , true , profileLocal . getTargetTime ( plasmaSearchProfile . PROCESS_COLLECTION ) ) ;
profileLocal . setYieldTime ( plasmaSearchProfile . PROCESS_COLLECTION ) ;
profileLocal . setYieldCount ( plasmaSearchProfile . PROCESS_COLLECTION , ( entities = = null ) ? 0 : entities . size ( ) ) ;
2005-10-12 14:28:49 +02:00
// since this is a conjunction we return an empty entity if any word is not known
2005-10-13 15:57:15 +02:00
if ( entities = = null ) {
rcLocal = new plasmaWordIndexEntity ( null ) ;
return 0 ;
}
2005-10-12 14:28:49 +02:00
// join the result
2005-10-23 19:50:27 +02:00
profileLocal . startTimer ( ) ;
rcLocal = plasmaWordIndexEntity . joinEntities ( entities , profileLocal . getTargetTime ( plasmaSearchProfile . PROCESS_JOIN ) ) ;
profileLocal . setYieldTime ( plasmaSearchProfile . PROCESS_JOIN ) ;
profileLocal . setYieldCount ( plasmaSearchProfile . PROCESS_JOIN , rcLocal . size ( ) ) ;
2005-10-13 15:57:15 +02:00
return rcLocal . size ( ) ;
2005-10-10 02:32:15 +02:00
}
2005-10-23 19:50:27 +02:00
public int globalSearch ( int fetchpeers ) {
2005-10-13 15:57:15 +02:00
// do global fetching
// the result of the fetch is then in the rcGlobal
if ( fetchpeers < 10 ) fetchpeers = 10 ;
2005-10-23 19:50:27 +02:00
log . logFine ( " STARTING " + fetchpeers + " THREADS TO CATCH EACH " + profileGlobal . getTargetCount ( plasmaSearchProfile . PROCESS_POSTSORT ) + " URLs WITHIN " + ( profileGlobal . duetime ( ) / 1000 ) + " SECONDS " ) ;
2005-10-13 15:57:15 +02:00
2005-10-23 19:50:27 +02:00
long timeout = System . currentTimeMillis ( ) + profileGlobal . duetime ( ) + 4000 ;
searchThreads = yacySearch . searchHashes ( query . queryHashes , urlStore , rcGlobal , fetchpeers , plasmaSwitchboard . urlBlacklist , snippetCache , profileGlobal ) ;
2005-10-13 15:57:15 +02:00
// wait until wanted delay passed or wanted result appeared
while ( System . currentTimeMillis ( ) < timeout ) {
// check if all threads have been finished or results so far are enough
2005-10-23 19:50:27 +02:00
if ( rcGlobal . size ( ) > = profileGlobal . getTargetCount ( plasmaSearchProfile . PROCESS_POSTSORT ) * 3 ) break ; // we have enough
2005-10-13 15:57:15 +02:00
if ( yacySearch . remainingWaiting ( searchThreads ) = = 0 ) break ; // we cannot expect more
// wait a little time ..
try { Thread . currentThread ( ) . sleep ( 100 ) ; } catch ( InterruptedException e ) { }
}
return rcGlobal . size ( ) ;
}
2005-10-23 19:50:27 +02:00
public plasmaSearchResult order ( ) throws IOException {
2005-10-13 15:57:15 +02:00
// we collect the urlhashes and construct a list with urlEntry objects
// attention: if minEntries is too high, this method will not terminate within the maxTime
2005-10-12 14:28:49 +02:00
2005-10-13 15:57:15 +02:00
plasmaWordIndexEntity searchResult = new plasmaWordIndexEntity ( null ) ;
searchResult . merge ( rcLocal , - 1 ) ;
searchResult . merge ( rcGlobal , - 1 ) ;
2005-10-23 19:50:27 +02:00
long preorderTime = profileLocal . getTargetTime ( plasmaSearchProfile . PROCESS_PRESORT ) ;
long postorderTime = profileLocal . getTargetTime ( plasmaSearchProfile . PROCESS_POSTSORT ) ;
profileLocal . startTimer ( ) ;
plasmaSearchPreOrder preorder = new plasmaSearchPreOrder ( query ) ;
preorder . addEntity ( searchResult , preorderTime ) ;
profileLocal . setYieldTime ( plasmaSearchProfile . PROCESS_PRESORT ) ;
profileLocal . setYieldCount ( plasmaSearchProfile . PROCESS_PRESORT , rcLocal . size ( ) ) ;
profileLocal . startTimer ( ) ;
2005-10-12 14:28:49 +02:00
plasmaSearchResult acc = new plasmaSearchResult ( query ) ;
if ( searchResult = = null ) return acc ; // strange case where searchResult is not proper: acc is then empty
if ( searchResult . size ( ) = = 0 ) return acc ; // case that we have nothing to do
2005-10-23 19:50:27 +02:00
// start url-fetch
2005-10-12 14:28:49 +02:00
plasmaWordIndexEntry entry ;
2005-10-23 19:50:27 +02:00
long postorderLimitTime = ( postorderTime < 0 ) ? Long . MAX_VALUE : System . currentTimeMillis ( ) + postorderTime ;
2005-10-12 14:28:49 +02:00
plasmaCrawlLURL . Entry page ;
2005-10-23 19:50:27 +02:00
int minEntries = profileLocal . getTargetCount ( plasmaSearchProfile . PROCESS_POSTSORT ) ;
2005-10-12 14:28:49 +02:00
try {
2005-10-23 19:50:27 +02:00
while ( preorder . hasNext ( ) ) {
if ( ( acc . sizeFetched ( ) > = minEntries ) & & ( System . currentTimeMillis ( ) > = postorderLimitTime ) ) break ;
entry = ( plasmaWordIndexEntry ) preorder . next ( ) ;
2005-10-12 14:28:49 +02:00
// find the url entry
page = urlStore . getEntry ( entry . getUrlHash ( ) ) ;
// add a result
acc . addResult ( entry , page ) ;
}
} catch ( kelondroException ee ) {
serverLog . logSevere ( " PLASMA " , " Database Failure during plasmaSearch.order: " + ee . getMessage ( ) , ee ) ;
}
2005-10-23 19:50:27 +02:00
profileLocal . setYieldTime ( plasmaSearchProfile . PROCESS_URLFETCH ) ;
profileLocal . setYieldCount ( plasmaSearchProfile . PROCESS_URLFETCH , acc . sizeFetched ( ) ) ;
// start postsorting
profileLocal . startTimer ( ) ;
2005-10-12 14:28:49 +02:00
acc . sortResults ( ) ;
2005-10-23 19:50:27 +02:00
profileLocal . setYieldTime ( plasmaSearchProfile . PROCESS_POSTSORT ) ;
profileLocal . setYieldCount ( plasmaSearchProfile . PROCESS_POSTSORT , acc . sizeOrdered ( ) ) ;
return acc ;
2005-10-12 14:28:49 +02:00
}
2005-10-13 15:57:15 +02:00
public void flushResults ( ) {
// put all new results into wordIndex
// this must be called after search results had been computed
// it is wise to call this within a separate thread because this method waits untill all
if ( searchThreads = = null ) return ;
2005-10-23 19:50:27 +02:00
// wait until all threads are finished
2005-10-13 15:57:15 +02:00
int remaining ;
2005-10-23 19:50:27 +02:00
int count = 0 ;
String wordHash ;
2005-10-13 15:57:15 +02:00
long starttime = System . currentTimeMillis ( ) ;
while ( ( remaining = yacySearch . remainingWaiting ( searchThreads ) ) > 0 ) {
2005-10-23 19:50:27 +02:00
// flush the rcGlobal as much as is there so far
synchronized ( rcGlobal ) {
Iterator hashi = query . queryHashes . iterator ( ) ;
while ( hashi . hasNext ( ) ) {
wordHash = ( String ) hashi . next ( ) ;
Iterator i = rcGlobal . elements ( true ) ;
plasmaWordIndexEntry entry ;
while ( i . hasNext ( ) ) {
entry = ( plasmaWordIndexEntry ) i . next ( ) ;
wordIndex . addEntries ( plasmaWordIndexEntryContainer . instantContainer ( wordHash , System . currentTimeMillis ( ) , entry ) , false ) ;
}
}
// the rcGlobal was flushed, empty it
count + = rcGlobal . size ( ) ;
rcGlobal . deleteComplete ( ) ;
}
// wait a little bit before trying again
try { Thread . currentThread ( ) . sleep ( 3000 ) ; } catch ( InterruptedException e ) { }
2005-10-13 15:57:15 +02:00
if ( System . currentTimeMillis ( ) - starttime > 90000 ) {
yacySearch . interruptAlive ( searchThreads ) ;
2005-10-23 19:50:27 +02:00
serverLog . logFine ( " PLASMA " , " SEARCH FLUSH: " + remaining + " PEERS STILL BUSY; ABANDONED; SEARCH WAS " + query . queryWords ) ;
2005-10-13 15:57:15 +02:00
break ;
}
}
2005-10-23 19:50:27 +02:00
serverLog . logFine ( " PLASMA " , " FINISHED FLUSHING " + count + " GLOBAL SEARCH RESULTS FOR SEARCH " + query . queryWords ) ;
2005-10-13 15:57:15 +02:00
// finally delete the temporary index
rcGlobal = null ;
}
2005-10-12 14:28:49 +02:00
/ *
public void preSearch ( ) {
plasmaWordIndexEntity idx = null ;
try {
// search the database locally
log . logFine ( " presearch: started job " ) ;
idx = searchHashes ( query . queryHashes , time ) ;
log . logFine ( " presearch: found " + idx . size ( ) + " results " ) ;
plasmaSearchResult acc = order ( idx , queryhashes , order , time , searchcount ) ;
if ( acc = = null ) return ;
log . logFine ( " presearch: ordered results, now " + acc . sizeOrdered ( ) + " URLs ready for fetch " ) ;
// take some elements and fetch the snippets
snippetCache . fetch ( acc , queryhashes , urlmask , fetchcount ) ;
} catch ( IOException e ) {
log . logSevere ( " presearch: failed " , e ) ;
} finally {
if ( idx ! = null ) try { idx . close ( ) ; } catch ( Exception e ) { }
}
log . logFine ( " presearch: job terminated " ) ;
}
* /
2005-10-10 02:32:15 +02:00
}