2005-09-25 03:09:21 +02:00
// plasmaCrawlBalancer.java
// -----------------------
// part of YaCy
2008-07-20 19:14:51 +02:00
// (C) by Michael Peter Christen; mc@yacy.net
2005-09-25 03:09:21 +02:00
// first published on http://www.anomic.de
// Frankfurt, Germany, 2005
// created: 24.09.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
2008-05-06 02:32:41 +02:00
package de.anomic.crawler ;
2005-09-25 03:09:21 +02:00
import java.io.File ;
import java.io.IOException ;
2006-09-30 00:27:20 +02:00
import java.util.ArrayList ;
2007-02-21 17:23:31 +01:00
import java.util.Collections ;
2005-09-25 03:09:21 +02:00
import java.util.HashMap ;
2008-04-21 00:10:26 +02:00
import java.util.HashSet ;
2005-09-25 03:09:21 +02:00
import java.util.Iterator ;
2007-02-21 17:23:31 +01:00
import java.util.LinkedList ;
2006-09-30 00:27:20 +02:00
import java.util.Map ;
2007-02-21 17:23:31 +01:00
import java.util.TreeMap ;
2005-09-25 03:09:21 +02:00
2006-12-06 04:02:57 +01:00
import de.anomic.kelondro.kelondroBase64Order ;
2008-01-17 22:48:08 +01:00
import de.anomic.kelondro.kelondroEcoTable ;
2007-03-16 14:25:56 +01:00
import de.anomic.kelondro.kelondroIndex ;
2006-06-02 14:45:57 +02:00
import de.anomic.kelondro.kelondroRow ;
2005-09-25 03:09:21 +02:00
import de.anomic.kelondro.kelondroStack ;
2008-05-06 02:32:41 +02:00
import de.anomic.plasma.plasmaSwitchboard ;
2007-02-21 17:23:31 +01:00
import de.anomic.server.logging.serverLog ;
2006-11-08 17:17:47 +01:00
import de.anomic.yacy.yacySeedDB ;
2005-09-25 03:09:21 +02:00
2008-05-06 02:32:41 +02:00
public class Balancer {
2007-03-16 14:25:56 +01:00
2008-01-17 22:48:08 +01:00
private static final String stackSuffix = " 9.stack " ;
private static final String indexSuffix = " 9.db " ;
private static final int EcoFSBufferSize = 200 ;
2005-09-25 03:09:21 +02:00
2007-02-21 17:23:31 +01:00
// a shared domainAccess map for all balancers
2008-01-07 23:36:48 +01:00
private static final Map < String , domaccess > domainAccess = Collections . synchronizedMap ( new HashMap < String , domaccess > ( ) ) ;
2007-02-21 17:23:31 +01:00
// definition of payload for fileStack
2007-03-16 14:25:56 +01:00
private static final kelondroRow stackrow = new kelondroRow ( " byte[] urlhash- " + yacySeedDB . commonHashLength , kelondroBase64Order . enhancedCoder , 0 ) ;
2007-02-21 17:23:31 +01:00
// class variables
2008-08-02 14:12:04 +02:00
private final ArrayList < String > urlRAMStack ; // a list that is flushed first
2008-01-07 23:36:48 +01:00
private kelondroStack urlFileStack ; // a file with url hashes
2008-03-31 20:45:27 +02:00
kelondroIndex urlFileIndex ;
2008-08-02 14:12:04 +02:00
private final HashMap < String , LinkedList < String > > domainStacks ; // a map from domain name part to Lists with url hashs
private final File cacheStacksPath ;
private final String stackname ;
2008-01-07 23:36:48 +01:00
private boolean top ; // to alternate between top and bottom of the file stack
2008-08-02 14:12:04 +02:00
private final boolean fullram ;
2008-01-07 23:36:48 +01:00
public static class domaccess {
long time ;
int count ;
public domaccess ( ) {
this . time = System . currentTimeMillis ( ) ;
this . count = 0 ;
}
public void update ( ) {
this . time = System . currentTimeMillis ( ) ;
this . count + + ;
}
public long time ( ) {
return this . time ;
}
public int count ( ) {
return this . count ;
}
}
2005-09-25 03:09:21 +02:00
2008-08-02 14:12:04 +02:00
public Balancer ( final File cachePath , final String stackname , final boolean fullram ) {
2007-03-16 14:25:56 +01:00
this . cacheStacksPath = cachePath ;
this . stackname = stackname ;
2008-08-02 14:12:04 +02:00
final File stackFile = new File ( cachePath , stackname + stackSuffix ) ;
2007-09-28 03:21:31 +02:00
this . urlFileStack = kelondroStack . open ( stackFile , stackrow ) ;
2008-01-07 23:36:48 +01:00
this . domainStacks = new HashMap < String , LinkedList < String > > ( ) ;
this . urlRAMStack = new ArrayList < String > ( ) ;
2007-09-28 03:21:31 +02:00
this . top = true ;
2008-01-19 02:50:24 +01:00
this . fullram = fullram ;
2007-03-16 14:25:56 +01:00
// create a stack for newly entered entries
if ( ! ( cachePath . exists ( ) ) ) cachePath . mkdir ( ) ; // make the path
openFileIndex ( ) ;
2008-07-03 15:08:37 +02:00
if ( urlFileStack . size ( ) ! = urlFileIndex . size ( ) | | ( urlFileIndex . size ( ) < 10000 & & urlFileIndex . size ( ) > 0 ) ) {
// fix the file stack
serverLog . logInfo ( " Balancer " , " re-creating the " + stackname + " balancer stack, size = " + urlFileIndex . size ( ) + ( ( urlFileStack . size ( ) = = urlFileIndex . size ( ) ) ? " " : " (the old stack size was wrong) " ) ) ;
urlFileStack = kelondroStack . reset ( urlFileStack ) ;
try {
2008-08-02 14:12:04 +02:00
final Iterator < byte [ ] > i = urlFileIndex . keys ( true , null ) ;
2008-07-03 15:08:37 +02:00
byte [ ] hash ;
while ( i . hasNext ( ) ) {
hash = i . next ( ) ;
pushHash ( new String ( hash ) ) ;
}
2008-08-02 14:12:04 +02:00
} catch ( final IOException e ) {
2008-07-03 15:08:37 +02:00
e . printStackTrace ( ) ;
}
}
2005-09-25 03:09:21 +02:00
}
2007-02-21 17:23:31 +01:00
public synchronized void close ( ) {
2007-08-03 14:21:46 +02:00
while ( domainStacksNotEmpty ( ) ) flushOnceDomStacks ( 0 , true ) ; // flush to ram, because the ram flush is optimized
2008-03-15 01:49:16 +01:00
size ( ) ;
2008-08-02 14:12:04 +02:00
try { flushAllRamStack ( ) ; } catch ( final IOException e ) { }
2007-03-16 14:25:56 +01:00
if ( urlFileIndex ! = null ) {
urlFileIndex . close ( ) ;
urlFileIndex = null ;
}
if ( urlFileStack ! = null ) {
urlFileStack . close ( ) ;
urlFileStack = null ;
}
2005-09-25 03:09:21 +02:00
}
2007-02-21 17:23:31 +01:00
public void finalize ( ) {
2008-02-25 15:08:15 +01:00
if ( urlFileStack ! = null ) {
2008-07-03 15:08:37 +02:00
serverLog . logWarning ( " Balancer " , " crawl stack " + stackname + " closed by finalizer " ) ;
2008-02-25 15:08:15 +01:00
close ( ) ;
}
2007-02-21 17:23:31 +01:00
}
public synchronized void clear ( ) {
2007-03-16 14:25:56 +01:00
urlFileStack = kelondroStack . reset ( urlFileStack ) ;
2007-02-21 17:23:31 +01:00
domainStacks . clear ( ) ;
2007-03-16 14:25:56 +01:00
urlRAMStack . clear ( ) ;
resetFileIndex ( ) ;
2007-02-21 17:23:31 +01:00
}
2007-03-16 14:25:56 +01:00
private void openFileIndex ( ) {
cacheStacksPath . mkdirs ( ) ;
2008-05-06 02:32:41 +02:00
urlFileIndex = new kelondroEcoTable ( new File ( cacheStacksPath , stackname + indexSuffix ) , CrawlEntry . rowdef , ( fullram ) ? kelondroEcoTable . tailCacheUsageAuto : kelondroEcoTable . tailCacheDenyUsage , EcoFSBufferSize , 0 ) ;
2007-02-21 17:23:31 +01:00
}
2007-03-16 14:25:56 +01:00
private void resetFileIndex ( ) {
if ( urlFileIndex ! = null ) {
urlFileIndex . close ( ) ;
urlFileIndex = null ;
2008-01-17 22:48:08 +01:00
new File ( cacheStacksPath , stackname + indexSuffix ) . delete ( ) ;
2007-03-16 14:25:56 +01:00
}
openFileIndex ( ) ;
2005-10-05 02:52:55 +02:00
}
2008-08-02 14:12:04 +02:00
public synchronized CrawlEntry get ( final String urlhash ) throws IOException {
2008-02-15 14:12:25 +01:00
assert urlhash ! = null ;
2008-07-11 12:59:06 +02:00
if ( urlFileIndex = = null ) return null ; // case occurs during shutdown
2008-08-02 14:12:04 +02:00
final kelondroRow . Entry entry = urlFileIndex . get ( urlhash . getBytes ( ) ) ;
2008-07-04 01:37:04 +02:00
if ( entry = = null ) return null ;
return new CrawlEntry ( entry ) ;
2007-03-16 14:25:56 +01:00
}
2008-08-02 14:12:04 +02:00
public synchronized int removeAllByProfileHandle ( final String profileHandle , final long timeout ) throws IOException {
2007-09-28 03:21:31 +02:00
// removes all entries with a specific profile hash.
// this may last some time
// returns number of deletions
// first find a list of url hashes that shall be deleted
2008-08-02 14:12:04 +02:00
final Iterator < kelondroRow . Entry > i = urlFileIndex . rows ( true , null ) ;
final HashSet < String > urlHashes = new HashSet < String > ( ) ;
2007-09-28 03:21:31 +02:00
kelondroRow . Entry rowEntry ;
2008-05-06 02:32:41 +02:00
CrawlEntry crawlEntry ;
2008-08-02 14:12:04 +02:00
final long terminate = ( timeout > 0 ) ? System . currentTimeMillis ( ) + timeout : Long . MAX_VALUE ;
2008-05-12 16:58:56 +02:00
while ( i . hasNext ( ) & & ( System . currentTimeMillis ( ) < terminate ) ) {
2008-06-06 18:01:27 +02:00
rowEntry = i . next ( ) ;
2008-05-06 02:32:41 +02:00
crawlEntry = new CrawlEntry ( rowEntry ) ;
2007-09-28 03:21:31 +02:00
if ( crawlEntry . profileHandle ( ) . equals ( profileHandle ) ) {
urlHashes . add ( crawlEntry . url ( ) . hash ( ) ) ;
}
}
// then delete all these urls from the queues and the file index
2008-04-21 00:10:26 +02:00
return this . remove ( urlHashes ) ;
2007-09-28 03:21:31 +02:00
}
2008-04-21 00:10:26 +02:00
/ * *
* this method is only here , because so many import / export methods need it
and it was implemented in the previous architecture
however , usage is not recommended
* @param urlHashes , a list of hashes that shall be removed
* @return number of entries that had been removed
* @throws IOException
* /
2008-08-02 14:12:04 +02:00
public synchronized int remove ( final HashSet < String > urlHashes ) throws IOException {
final int s = urlFileIndex . size ( ) ;
2008-04-21 00:10:26 +02:00
int removedCounter = 0 ;
2008-08-02 14:12:04 +02:00
for ( final String urlhash : urlHashes ) {
final kelondroRow . Entry entry = urlFileIndex . remove ( urlhash . getBytes ( ) ) ;
2008-04-21 00:10:26 +02:00
if ( entry ! = null ) removedCounter + + ;
}
if ( removedCounter = = 0 ) return 0 ;
assert urlFileIndex . size ( ) + removedCounter = = s : " urlFileIndex.size() = " + urlFileIndex . size ( ) + " , s = " + s ;
2007-03-16 14:25:56 +01:00
2008-04-21 00:10:26 +02:00
// now delete these hashes also from the queues
2007-04-17 17:15:47 +02:00
2008-04-21 00:10:26 +02:00
// iterate through the RAM stack
Iterator < String > i = urlRAMStack . iterator ( ) ;
String h ;
while ( i . hasNext ( ) ) {
2008-06-06 18:01:27 +02:00
h = i . next ( ) ;
2008-04-21 00:10:26 +02:00
if ( urlHashes . contains ( h ) ) i . remove ( ) ;
2007-03-16 14:25:56 +01:00
}
2007-04-17 17:15:47 +02:00
// iterate through the file stack
// in general this is a bad idea. But this can only be avoided by avoidance of this method
2008-08-02 14:12:04 +02:00
final Iterator < kelondroRow . Entry > j = urlFileStack . stackIterator ( true ) ;
2008-01-07 23:36:48 +01:00
while ( j . hasNext ( ) ) {
h = new String ( j . next ( ) . getColBytes ( 0 ) ) ;
2008-04-21 00:10:26 +02:00
if ( urlHashes . contains ( h ) ) j . remove ( ) ;
2007-04-17 17:15:47 +02:00
}
2007-03-16 14:25:56 +01:00
2008-07-11 14:03:18 +02:00
// iterate through the domain stacks
2008-08-02 14:12:04 +02:00
final Iterator < Map . Entry < String , LinkedList < String > > > k = domainStacks . entrySet ( ) . iterator ( ) ;
2008-07-11 14:03:18 +02:00
Map . Entry < String , LinkedList < String > > se ;
LinkedList < String > stack ;
while ( k . hasNext ( ) ) {
se = k . next ( ) ;
stack = se . getValue ( ) ;
i = stack . iterator ( ) ;
while ( i . hasNext ( ) ) {
if ( urlHashes . contains ( i . next ( ) ) ) i . remove ( ) ;
}
if ( stack . size ( ) = = 0 ) k . remove ( ) ;
}
2008-04-21 00:10:26 +02:00
return removedCounter ;
2007-03-16 14:25:56 +01:00
}
2008-08-02 14:12:04 +02:00
public synchronized boolean has ( final String urlhash ) {
2008-06-18 01:56:39 +02:00
return urlFileIndex . has ( urlhash . getBytes ( ) ) ;
2005-09-25 03:09:21 +02:00
}
2007-08-03 14:21:46 +02:00
public boolean notEmpty ( ) {
// alternative method to the property size() > 0
// this is better because it may avoid synchronized access to domain stack summarization
return urlRAMStack . size ( ) > 0 | | urlFileStack . size ( ) > 0 | | domainStacksNotEmpty ( ) ;
}
2008-07-11 11:12:54 +02:00
public int size ( ) {
2008-08-02 14:12:04 +02:00
final int componentsize = urlFileIndex . size ( ) ;
2008-07-11 14:03:18 +02:00
/ *
2008-07-11 11:12:54 +02:00
assert componentsize = = urlFileStack . size ( ) + urlRAMStack . size ( ) + sizeDomainStacks ( ) :
" size wrong in " + stackname +
" - urlFileIndex = " + urlFileIndex . size ( ) +
2008-07-11 14:03:18 +02:00
" , componentsize = " + urlFileStack . size ( ) + urlRAMStack . size ( ) + sizeDomainStacks ( ) +
2008-07-11 11:12:54 +02:00
" = (urlFileStack = " + urlFileStack . size ( ) +
" , urlRAMStack = " + urlRAMStack . size ( ) +
" , sizeDomainStacks = " + sizeDomainStacks ( ) + " ) " ;
2008-07-11 14:03:18 +02:00
* /
2007-02-21 17:23:31 +01:00
return componentsize ;
2005-09-25 03:09:21 +02:00
}
2007-08-03 14:21:46 +02:00
private boolean domainStacksNotEmpty ( ) {
if ( domainStacks = = null ) return false ;
synchronized ( domainStacks ) {
2008-08-02 14:12:04 +02:00
final Iterator < LinkedList < String > > i = domainStacks . values ( ) . iterator ( ) ;
2007-08-03 14:21:46 +02:00
while ( i . hasNext ( ) ) {
2008-01-07 23:36:48 +01:00
if ( i . next ( ) . size ( ) > 0 ) return true ;
2007-08-03 14:21:46 +02:00
}
}
return false ;
}
2005-09-25 03:09:21 +02:00
private int sizeDomainStacks ( ) {
2006-07-13 03:31:00 +02:00
if ( domainStacks = = null ) return 0 ;
2005-09-25 03:09:21 +02:00
int sum = 0 ;
2007-08-03 13:44:58 +02:00
synchronized ( domainStacks ) {
2008-08-02 14:12:04 +02:00
final Iterator < LinkedList < String > > i = domainStacks . values ( ) . iterator ( ) ;
2008-01-07 23:36:48 +01:00
while ( i . hasNext ( ) ) sum + = i . next ( ) . size ( ) ;
2007-08-03 13:44:58 +02:00
}
2005-09-25 03:09:21 +02:00
return sum ;
}
2008-08-02 14:12:04 +02:00
private void flushOnceDomStacks ( final int minimumleft , final boolean ram ) {
2007-03-16 17:54:54 +01:00
// takes one entry from every domain stack and puts it on the ram or file stack
// the minimumleft value is a limit for the number of entries that should be left
2007-02-21 17:23:31 +01:00
if ( domainStacks . size ( ) = = 0 ) return ;
2007-08-03 14:21:46 +02:00
synchronized ( domainStacks ) {
2008-08-02 14:12:04 +02:00
final Iterator < Map . Entry < String , LinkedList < String > > > i = domainStacks . entrySet ( ) . iterator ( ) ;
2008-01-07 23:36:48 +01:00
Map . Entry < String , LinkedList < String > > entry ;
LinkedList < String > list ;
2007-08-03 14:21:46 +02:00
while ( i . hasNext ( ) ) {
2008-01-07 23:36:48 +01:00
entry = i . next ( ) ;
list = entry . getValue ( ) ;
2007-08-03 14:21:46 +02:00
if ( list . size ( ) > minimumleft ) {
if ( ram ) {
urlRAMStack . add ( list . removeFirst ( ) ) ;
} else try {
2008-06-06 18:01:27 +02:00
urlFileStack . push ( urlFileStack . row ( ) . newEntry ( new byte [ ] [ ] { ( list . removeFirst ( ) ) . getBytes ( ) } ) ) ;
2008-08-02 14:12:04 +02:00
} catch ( final IOException e ) {
2007-08-03 14:21:46 +02:00
e . printStackTrace ( ) ;
}
2007-02-20 09:35:51 +01:00
}
2007-08-03 14:21:46 +02:00
if ( list . size ( ) = = 0 ) i . remove ( ) ;
2005-09-25 03:09:21 +02:00
}
}
}
2007-02-21 17:23:31 +01:00
private void flushAllRamStack ( ) throws IOException {
// this flushes only the ramStack to the fileStack, but does not flush the domainStacks
2007-03-16 14:25:56 +01:00
for ( int i = 0 ; i < urlRAMStack . size ( ) / 2 ; i + + ) {
2008-06-06 18:01:27 +02:00
urlFileStack . push ( urlFileStack . row ( ) . newEntry ( new byte [ ] [ ] { ( urlRAMStack . get ( i ) ) . getBytes ( ) } ) ) ;
urlFileStack . push ( urlFileStack . row ( ) . newEntry ( new byte [ ] [ ] { ( urlRAMStack . get ( urlRAMStack . size ( ) - i - 1 ) ) . getBytes ( ) } ) ) ;
2007-02-21 17:23:31 +01:00
}
2007-03-16 14:25:56 +01:00
if ( urlRAMStack . size ( ) % 2 = = 1 )
2008-06-06 18:01:27 +02:00
urlFileStack . push ( urlFileStack . row ( ) . newEntry ( new byte [ ] [ ] { ( urlRAMStack . get ( urlRAMStack . size ( ) / 2 ) ) . getBytes ( ) } ) ) ;
2005-09-25 03:09:21 +02:00
}
2008-08-02 14:12:04 +02:00
public synchronized void push ( final CrawlEntry entry ) throws IOException {
2007-03-16 14:25:56 +01:00
assert entry ! = null ;
2007-09-05 11:01:35 +02:00
if ( urlFileIndex . has ( entry . url ( ) . hash ( ) . getBytes ( ) ) ) {
serverLog . logWarning ( " PLASMA BALANCER " , " double-check has failed for urlhash " + entry . url ( ) . hash ( ) + " in " + stackname + " - fixed " ) ;
2007-02-21 17:23:31 +01:00
return ;
}
2007-03-16 14:25:56 +01:00
2008-07-03 15:08:37 +02:00
// add to index
urlFileIndex . put ( entry . toRow ( ) ) ;
// add the hash to a queue
pushHash ( entry . url ( ) . hash ( ) ) ;
}
2008-08-02 14:12:04 +02:00
private void pushHash ( final String hash ) throws IOException {
2007-03-16 14:25:56 +01:00
// extend domain stack
2008-08-02 14:12:04 +02:00
final String dom = hash . substring ( 6 ) ;
2008-01-07 23:36:48 +01:00
LinkedList < String > domainList = domainStacks . get ( dom ) ;
2007-02-21 17:23:31 +01:00
if ( domainList = = null ) {
// create new list
2008-01-07 23:36:48 +01:00
domainList = new LinkedList < String > ( ) ;
2007-08-03 14:21:46 +02:00
synchronized ( domainStacks ) {
2008-07-03 15:08:37 +02:00
domainList . add ( hash ) ;
2007-08-03 14:21:46 +02:00
domainStacks . put ( dom , domainList ) ;
}
2007-02-21 17:23:31 +01:00
} else {
// extend existent domain list
2008-07-03 15:08:37 +02:00
domainList . addLast ( hash ) ;
2005-09-25 03:09:21 +02:00
}
// check size of domainStacks and flush
2008-06-18 01:56:39 +02:00
if ( ( domainStacks . size ( ) > 100 ) | | ( sizeDomainStacks ( ) > 1000 ) ) {
2007-03-16 17:54:54 +01:00
flushOnceDomStacks ( 1 , urlRAMStack . size ( ) < 100 ) ; // when the ram stack is small, flush it there
2005-09-25 03:09:21 +02:00
}
}
2008-08-02 14:12:04 +02:00
public synchronized CrawlEntry pop ( final long minimumLocalDelta , final long minimumGlobalDelta , final long maximumAge ) throws IOException {
2007-02-09 11:32:58 +01:00
// returns an url-hash from the stack and ensures minimum delta times
2007-02-21 17:23:31 +01:00
// we have 3 sources to choose from: the ramStack, the domainStacks and the fileStack
String result = null ; // the result
// 1st: check ramStack
2007-03-16 14:25:56 +01:00
if ( urlRAMStack . size ( ) > 0 ) {
2008-06-06 18:01:27 +02:00
result = urlRAMStack . remove ( 0 ) ;
2007-02-21 17:23:31 +01:00
}
// 2nd-a: check domainStacks for latest arrivals
2007-08-03 14:21:46 +02:00
if ( ( result = = null ) & & ( domainStacks . size ( ) > 0 ) ) synchronized ( domainStacks ) {
2007-02-21 17:23:31 +01:00
// we select specific domains that have not been used for a long time
// i.e. 60 seconds. Latest arrivals that have not yet been crawled
// fit also in that scheme
2008-08-02 14:12:04 +02:00
final Iterator < Map . Entry < String , LinkedList < String > > > i = domainStacks . entrySet ( ) . iterator ( ) ;
2008-01-07 23:36:48 +01:00
Map . Entry < String , LinkedList < String > > entry ;
2007-02-21 17:23:31 +01:00
String domhash ;
long delta , maxdelta = 0 ;
String maxhash = null ;
2008-01-07 23:36:48 +01:00
LinkedList < String > domlist ;
2007-02-21 17:23:31 +01:00
while ( i . hasNext ( ) ) {
2008-01-07 23:36:48 +01:00
entry = i . next ( ) ;
2008-06-06 18:01:27 +02:00
domhash = entry . getKey ( ) ;
2007-02-21 17:23:31 +01:00
delta = lastAccessDelta ( domhash ) ;
if ( delta = = Integer . MAX_VALUE ) {
// a brand new domain - we take it
2008-01-07 23:36:48 +01:00
domlist = entry . getValue ( ) ;
2008-06-06 18:01:27 +02:00
result = domlist . removeFirst ( ) ;
2007-02-21 17:23:31 +01:00
if ( domlist . size ( ) = = 0 ) i . remove ( ) ;
break ;
}
if ( delta > maxdelta ) {
maxdelta = delta ;
maxhash = domhash ;
}
}
if ( maxdelta > maximumAge ) {
// success - we found an entry from a domain that has not been used for a long time
2008-01-07 23:36:48 +01:00
domlist = domainStacks . get ( maxhash ) ;
2008-06-06 18:01:27 +02:00
result = domlist . removeFirst ( ) ;
2007-02-21 17:23:31 +01:00
if ( domlist . size ( ) = = 0 ) domainStacks . remove ( maxhash ) ;
}
}
// 2nd-b: check domainStacks for best match between stack size and retrieval time
2007-08-03 14:21:46 +02:00
if ( ( result = = null ) & & ( domainStacks . size ( ) > 0 ) ) synchronized ( domainStacks ) {
2007-02-21 17:23:31 +01:00
// we order all domains by the number of entries per domain
// then we iterate through these domains in descending entry order
// and that that one, that has a delta > minimumDelta
2008-08-02 14:12:04 +02:00
final Iterator < Map . Entry < String , LinkedList < String > > > i = domainStacks . entrySet ( ) . iterator ( ) ;
2008-01-07 23:36:48 +01:00
Map . Entry < String , LinkedList < String > > entry ;
2007-02-21 17:23:31 +01:00
String domhash ;
2008-01-07 23:36:48 +01:00
LinkedList < String > domlist ;
2008-08-02 14:12:04 +02:00
final TreeMap < Integer , String > hitlist = new TreeMap < Integer , String > ( ) ;
2007-02-21 17:23:31 +01:00
int count = 0 ;
// first collect information about sizes of the domain lists
while ( i . hasNext ( ) ) {
2008-01-07 23:36:48 +01:00
entry = i . next ( ) ;
domhash = entry . getKey ( ) ;
domlist = entry . getValue ( ) ;
2007-02-21 17:23:31 +01:00
hitlist . put ( new Integer ( domlist . size ( ) * 100 + count + + ) , domhash ) ;
}
2007-02-09 11:32:58 +01:00
2008-07-05 02:35:20 +02:00
// now iterate in descending order and fetch that one,
2007-02-21 17:23:31 +01:00
// that is acceptable by the minimumDelta constraint
long delta ;
String maxhash = null ;
while ( hitlist . size ( ) > 0 ) {
2008-06-06 18:01:27 +02:00
domhash = hitlist . remove ( hitlist . lastKey ( ) ) ;
2007-02-21 17:23:31 +01:00
if ( maxhash = = null ) maxhash = domhash ; // remember first entry
delta = lastAccessDelta ( domhash ) ;
2007-08-16 01:18:12 +02:00
if ( delta > minimumGlobalDelta ) {
2008-01-07 23:36:48 +01:00
domlist = domainStacks . get ( domhash ) ;
2008-06-06 18:01:27 +02:00
result = domlist . removeFirst ( ) ;
2007-02-21 17:23:31 +01:00
if ( domlist . size ( ) = = 0 ) domainStacks . remove ( domhash ) ;
break ;
}
2007-02-09 10:48:23 +01:00
}
2007-02-09 11:32:58 +01:00
2007-02-21 17:23:31 +01:00
// if we did yet not choose any entry, we simply take that one with the most entries
if ( ( result = = null ) & & ( maxhash ! = null ) ) {
2008-01-07 23:36:48 +01:00
domlist = domainStacks . get ( maxhash ) ;
2008-06-06 18:01:27 +02:00
result = domlist . removeFirst ( ) ;
2007-02-21 17:23:31 +01:00
if ( domlist . size ( ) = = 0 ) domainStacks . remove ( maxhash ) ;
2006-07-13 03:31:00 +02:00
}
}
2005-09-25 03:09:21 +02:00
2007-02-21 17:23:31 +01:00
// 3rd: take entry from file
2007-03-16 14:25:56 +01:00
if ( ( result = = null ) & & ( urlFileStack . size ( ) > 0 ) ) {
2008-08-02 14:12:04 +02:00
final kelondroRow . Entry nextentry = ( top ) ? urlFileStack . top ( ) : urlFileStack . bot ( ) ;
2007-03-16 17:54:54 +01:00
if ( nextentry = = null ) {
2007-03-07 10:08:13 +01:00
// emergency case: this means that something with the stack organization is wrong
// the file appears to be broken. We kill the file.
2007-03-16 14:25:56 +01:00
kelondroStack . reset ( urlFileStack ) ;
2007-03-07 10:08:13 +01:00
serverLog . logSevere ( " PLASMA BALANCER " , " get() failed to fetch entry from file stack. reset stack file. " ) ;
} else {
2008-08-02 14:12:04 +02:00
final String nexthash = new String ( nextentry . getColBytes ( 0 ) ) ;
2007-02-21 17:23:31 +01:00
// check if the time after retrieval of last hash from same
// domain is not shorter than the minimumDelta
2007-03-16 17:54:54 +01:00
long delta = lastAccessDelta ( nexthash ) ;
2007-08-16 01:18:12 +02:00
if ( delta > minimumGlobalDelta ) {
2007-03-16 17:54:54 +01:00
// the entry is fine
result = new String ( ( top ) ? urlFileStack . pop ( ) . getColBytes ( 0 ) : urlFileStack . pot ( ) . getColBytes ( 0 ) ) ;
2007-02-21 17:23:31 +01:00
} else {
2007-03-16 17:54:54 +01:00
// try other entry
result = new String ( ( top ) ? urlFileStack . pot ( ) . getColBytes ( 0 ) : urlFileStack . pop ( ) . getColBytes ( 0 ) ) ;
2007-02-21 17:23:31 +01:00
delta = lastAccessDelta ( result ) ;
}
}
2007-03-16 17:54:54 +01:00
top = ! top ; // alternate top/bottom
2005-09-25 03:09:21 +02:00
}
2007-02-21 17:23:31 +01:00
// check case where we did not found anything
if ( result = = null ) {
2007-03-16 14:25:56 +01:00
serverLog . logSevere ( " PLASMA BALANCER " , " get() was not able to find a valid urlhash - total size = " + size ( ) + " , fileStack.size() = " + urlFileStack . size ( ) + " , ramStack.size() = " + urlRAMStack . size ( ) + " , domainStacks.size() = " + domainStacks . size ( ) ) ;
2007-02-21 17:23:31 +01:00
return null ;
2005-09-25 03:09:21 +02:00
}
2007-02-21 17:23:31 +01:00
// finally: check minimumDelta and if necessary force a sleep
2008-08-02 14:12:04 +02:00
final long delta = lastAccessDelta ( result ) ;
2007-04-30 02:39:53 +02:00
assert delta > = 0 : " delta = " + delta ;
2008-08-02 14:12:04 +02:00
final int s = urlFileIndex . size ( ) ;
final kelondroRow . Entry rowEntry = urlFileIndex . remove ( result . getBytes ( ) ) ;
2008-01-25 00:58:18 +01:00
assert ( rowEntry = = null ) | | ( urlFileIndex . size ( ) + 1 = = s ) : " urlFileIndex.size() = " + urlFileIndex . size ( ) + " , s = " + s + " , result = " + result ;
2007-06-06 09:53:56 +02:00
if ( rowEntry = = null ) {
serverLog . logSevere ( " PLASMA BALANCER " , " get() found a valid urlhash, but failed to fetch the corresponding url entry - total size = " + size ( ) + " , fileStack.size() = " + urlFileStack . size ( ) + " , ramStack.size() = " + urlRAMStack . size ( ) + " , domainStacks.size() = " + domainStacks . size ( ) ) ;
return null ;
2008-01-22 21:49:26 +01:00
} else {
assert urlFileIndex . size ( ) + 1 = = s : " urlFileIndex.size() = " + urlFileIndex . size ( ) + " , s = " + s + " , result = " + result ;
2007-06-06 09:53:56 +02:00
}
2008-08-02 14:12:04 +02:00
final CrawlEntry crawlEntry = new CrawlEntry ( rowEntry ) ;
final long genericDelta = Math . min (
2008-07-05 02:35:20 +02:00
15000 ,
Math . max (
( crawlEntry . url ( ) . isLocal ( ) ) ? minimumLocalDelta : minimumGlobalDelta ,
2008-07-11 11:12:54 +02:00
plasmaSwitchboard . getSwitchboard ( ) . robots . crawlDelay ( crawlEntry . url ( ) ) * 1000 )
2008-07-05 02:35:20 +02:00
) ; // prevent that that robots file can stop our indexer completely
2007-06-06 09:53:56 +02:00
if ( delta < genericDelta ) {
2007-02-21 17:23:31 +01:00
// force a busy waiting here
// in best case, this should never happen if the balancer works propertly
// this is only to protect against the worst case, where the crawler could
// behave in a DoS-manner
2008-08-02 14:12:04 +02:00
final long sleeptime = genericDelta - delta ;
try { synchronized ( this ) { this . wait ( sleeptime ) ; } } catch ( final InterruptedException e ) { }
2005-09-25 03:09:21 +02:00
}
2007-02-21 17:23:31 +01:00
// update statistical data
2008-01-07 23:36:48 +01:00
domaccess lastAccess = domainAccess . get ( result . substring ( 6 ) ) ;
if ( lastAccess = = null ) lastAccess = new domaccess ( ) ; else lastAccess . update ( ) ;
domainAccess . put ( result . substring ( 6 ) , lastAccess ) ;
2007-06-06 09:53:56 +02:00
return crawlEntry ;
2007-02-21 17:23:31 +01:00
}
2008-08-02 14:12:04 +02:00
private long lastAccessDelta ( final String hash ) {
2007-02-21 17:23:31 +01:00
assert hash ! = null ;
2008-08-02 14:12:04 +02:00
final domaccess lastAccess = domainAccess . get ( ( hash . length ( ) > 6 ) ? hash . substring ( 6 ) : hash ) ;
2007-02-21 17:23:31 +01:00
if ( lastAccess = = null ) return Long . MAX_VALUE ; // never accessed
2008-01-07 23:36:48 +01:00
return System . currentTimeMillis ( ) - lastAccess . time ( ) ;
2007-02-21 17:23:31 +01:00
}
2008-07-03 15:08:37 +02:00
public synchronized ArrayList < CrawlEntry > top ( int count ) throws IOException {
2007-03-16 16:16:26 +01:00
// if we need to flush anything, then flush the domain stack first,
// to avoid that new urls get hidden by old entries from the file stack
2007-09-09 19:31:29 +02:00
if ( urlRAMStack = = null ) return null ;
2008-07-03 15:08:37 +02:00
while ( ( domainStacksNotEmpty ( ) ) & & ( urlRAMStack . size ( ) < = count ) ) {
2007-03-16 16:16:26 +01:00
// flush only that much as we need to display
2007-03-16 17:54:54 +01:00
flushOnceDomStacks ( 0 , true ) ;
2007-03-16 16:16:26 +01:00
}
2008-07-03 15:08:37 +02:00
while ( ( urlFileStack ! = null ) & & ( urlRAMStack . size ( ) < = count ) & & ( urlFileStack . size ( ) > 0 ) ) {
2007-03-16 16:16:26 +01:00
// flush some entries from disc to ram stack
2007-02-21 17:23:31 +01:00
try {
2008-07-05 02:35:20 +02:00
// one from the top:
2007-08-27 00:06:58 +02:00
kelondroRow . Entry t = urlFileStack . pop ( ) ;
if ( t = = null ) break ;
urlRAMStack . add ( new String ( t . getColBytes ( 0 ) ) ) ;
2008-07-05 02:35:20 +02:00
if ( urlFileStack . size ( ) = = 0 ) break ;
// one from the bottom:
t = urlFileStack . pot ( ) ;
if ( t = = null ) break ;
urlRAMStack . add ( new String ( t . getColBytes ( 0 ) ) ) ;
2008-08-02 14:12:04 +02:00
} catch ( final IOException e ) {
2007-03-16 16:16:26 +01:00
break ;
}
2005-09-25 03:09:21 +02:00
}
2008-07-03 15:08:37 +02:00
count = Math . min ( count , urlRAMStack . size ( ) ) ;
2008-08-02 14:12:04 +02:00
final ArrayList < CrawlEntry > list = new ArrayList < CrawlEntry > ( ) ;
2008-07-03 15:08:37 +02:00
for ( int i = 0 ; i < count ; i + + ) {
2008-08-02 14:12:04 +02:00
final String urlhash = urlRAMStack . get ( i ) ;
final kelondroRow . Entry entry = urlFileIndex . get ( urlhash . getBytes ( ) ) ;
2008-07-03 15:08:37 +02:00
if ( entry = = null ) break ;
list . add ( new CrawlEntry ( entry ) ) ;
2007-03-16 16:16:26 +01:00
}
2008-07-03 15:08:37 +02:00
return list ;
2007-03-16 14:25:56 +01:00
}
2008-07-03 15:08:37 +02:00
2008-05-06 02:32:41 +02:00
public synchronized Iterator < CrawlEntry > iterator ( ) throws IOException {
2007-03-16 14:25:56 +01:00
return new EntryIterator ( ) ;
}
2008-05-06 02:32:41 +02:00
private class EntryIterator implements Iterator < CrawlEntry > {
2007-03-16 14:25:56 +01:00
2008-01-07 23:36:48 +01:00
private Iterator < kelondroRow . Entry > rowIterator ;
2007-03-16 14:25:56 +01:00
public EntryIterator ( ) throws IOException {
rowIterator = urlFileIndex . rows ( true , null ) ;
}
public boolean hasNext ( ) {
return ( rowIterator = = null ) ? false : rowIterator . hasNext ( ) ;
}
2008-05-06 02:32:41 +02:00
public CrawlEntry next ( ) {
2008-08-02 14:12:04 +02:00
final kelondroRow . Entry entry = rowIterator . next ( ) ;
2007-03-16 14:25:56 +01:00
try {
2008-05-06 02:32:41 +02:00
return ( entry = = null ) ? null : new CrawlEntry ( entry ) ;
2008-08-02 14:12:04 +02:00
} catch ( final IOException e ) {
2007-03-16 14:25:56 +01:00
rowIterator = null ;
return null ;
}
}
public void remove ( ) {
if ( rowIterator ! = null ) rowIterator . remove ( ) ;
}
2005-09-25 03:09:21 +02:00
}
}