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 ;
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 ;
2008-09-08 23:50:37 +02:00
import java.util.concurrent.ConcurrentHashMap ;
2005-09-25 03:09:21 +02:00
2009-01-30 16:33:00 +01:00
import de.anomic.kelondro.index.Row ;
import de.anomic.kelondro.index.ObjectIndex ;
2009-01-30 23:08:08 +01:00
import de.anomic.kelondro.order.Base64Order ;
import de.anomic.kelondro.table.EcoTable ;
2009-01-30 23:44:20 +01:00
import de.anomic.kelondro.table.Stack ;
2009-01-31 00:33:47 +01:00
import de.anomic.kelondro.util.Log ;
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
// definition of payload for fileStack
2009-01-30 16:33:00 +01:00
private static final Row stackrow = new Row ( " byte[] urlhash- " + yacySeedDB . commonHashLength , Base64Order . enhancedCoder , 0 ) ;
2007-02-21 17:23:31 +01:00
// class variables
2008-09-08 23:50:37 +02:00
private final ConcurrentHashMap < String , LinkedList < String > >
domainStacks ; // a map from domain name part to Lists with url hashs
private final ArrayList < String > urlRAMStack ; // a list that is flushed first
2009-01-30 23:44:20 +01:00
private Stack urlFileStack ; // a file with url hashes
2009-01-30 16:33:00 +01:00
private ObjectIndex urlFileIndex ;
2008-09-08 23:50:37 +02:00
private final File cacheStacksPath ;
private final String stackname ;
private boolean top ; // to alternate between top and bottom of the file stack
private final boolean fullram ;
2008-09-17 23:45:21 +02:00
private long minimumLocalDelta ;
private long minimumGlobalDelta ;
2005-09-25 03:09:21 +02:00
2008-09-17 23:45:21 +02:00
public Balancer ( final File cachePath , final String stackname , final boolean fullram ,
final long minimumLocalDelta , final long minimumGlobalDelta ) {
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 ) ;
2009-01-30 23:44:20 +01:00
this . urlFileStack = Stack . open ( stackFile , stackrow ) ;
2008-09-08 23:50:37 +02:00
this . domainStacks = new ConcurrentHashMap < String , LinkedList < String > > ( ) ;
2008-01-07 23:36:48 +01:00
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 ;
2008-09-17 23:45:21 +02:00
this . minimumLocalDelta = minimumLocalDelta ;
this . minimumGlobalDelta = minimumGlobalDelta ;
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
2009-01-31 00:33:47 +01:00
Log . logInfo ( " Balancer " , " re-creating the " + stackname + " balancer stack, size = " + urlFileIndex . size ( ) + ( ( urlFileStack . size ( ) = = urlFileIndex . size ( ) ) ? " " : " (the old stack size was wrong) " ) ) ;
2009-03-13 15:56:25 +01:00
urlFileStack . clear ( ) ;
2008-07-03 15:08:37 +02:00
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 ;
2009-03-13 00:05:18 +01:00
while ( i ! = null & & i . hasNext ( ) ) {
2008-07-03 15:08:37 +02:00
hash = i . next ( ) ;
2008-09-17 23:45:21 +02:00
pushHashToDomainStacks ( new String ( hash ) , true ) ;
2008-07-03 15:08:37 +02:00
}
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
}
2008-09-17 23:45:21 +02:00
public long getMinimumLocalDelta ( ) {
return this . minimumLocalDelta ;
}
public long getMinimumGlobalDelta ( ) {
return this . minimumGlobalDelta ;
}
public void setMinimumDelta ( final long minimumLocalDelta , final long minimumGlobalDelta ) {
this . minimumLocalDelta = minimumLocalDelta ;
this . minimumGlobalDelta = minimumGlobalDelta ;
}
2007-02-21 17:23:31 +01:00
public synchronized void close ( ) {
2008-11-22 12:25:01 +01:00
while ( domainStacksNotEmpty ( ) ) flushOnceDomStacks ( true , false , Integer . MAX_VALUE ) ; // 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
}
2008-08-06 21:43:12 +02:00
protected void finalize ( ) {
2008-02-25 15:08:15 +01:00
if ( urlFileStack ! = null ) {
2009-01-31 00:33:47 +01:00
Log . 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 ( ) {
2009-03-13 15:56:25 +01:00
urlFileStack . clear ( ) ;
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 ( ) ;
2009-01-30 23:08:08 +01:00
urlFileIndex = new EcoTable ( new File ( cacheStacksPath , stackname + indexSuffix ) , CrawlEntry . rowdef , ( fullram ) ? EcoTable . tailCacheUsageAuto : EcoTable . 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
2009-01-30 16:33:00 +01:00
final Row . 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
2009-02-24 11:40:20 +01:00
final Iterator < Row . Entry > i = urlFileIndex . rows ( ) ;
2008-08-02 14:12:04 +02:00
final HashSet < String > urlHashes = new HashSet < String > ( ) ;
2009-01-30 16:33:00 +01:00
Row . 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 ) {
2009-01-30 16:33:00 +01:00
final Row . 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
2009-01-30 16:33:00 +01:00
final Iterator < Row . 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
}
2009-02-17 10:12:47 +01: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
}
2009-02-17 10:12:47 +01:00
public synchronized boolean notEmpty ( ) {
2007-08-03 14:21:46 +02:00
// 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 ( ) ;
}
2009-02-17 10:12:47 +01:00
public synchronized 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 ;
2008-11-22 12:25:01 +01: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 ( ) ;
2008-11-22 12:25:01 +01:00
//}
2005-09-25 03:09:21 +02:00
return sum ;
}
2008-09-17 23:45:21 +02:00
/ * *
* removes the head element of all domain stacks and moves the element in either the ram stack or the file stack
* @param minimumleft
* @param ram
* @param onlyReadyForAccess
* /
2008-11-22 12:25:01 +01:00
private void flushOnceDomStacks ( final boolean ram , final boolean onlyReadyForAccess , int max ) {
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 ;
2008-11-22 12:25:01 +01:00
int c = 0 ;
while ( i . hasNext ( ) & & c < max ) {
2008-01-07 23:36:48 +01:00
entry = i . next ( ) ;
list = entry . getValue ( ) ;
2008-11-22 12:25:01 +01:00
if ( onlyReadyForAccess & & CrawlEntry . waitingRemainingGuessed ( list . getFirst ( ) , minimumLocalDelta , minimumGlobalDelta ) > 0 ) continue ;
if ( ram ) {
urlRAMStack . add ( list . removeFirst ( ) ) ;
} else try {
urlFileStack . push ( urlFileStack . row ( ) . newEntry ( new byte [ ] [ ] { ( list . removeFirst ( ) ) . getBytes ( ) } ) ) ;
} catch ( final IOException e ) {
e . printStackTrace ( ) ;
2007-02-20 09:35:51 +01:00
}
2007-08-03 14:21:46 +02:00
if ( list . size ( ) = = 0 ) i . remove ( ) ;
2008-11-22 12:25:01 +01:00
c + + ;
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
}
2008-08-06 21:43:12 +02:00
if ( urlRAMStack . size ( ) % 2 ! = 0 )
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-09-17 23:45:21 +02:00
private void shiftFileToDomStacks ( final int wantedsize ) {
int count = sizeDomainStacks ( ) - wantedsize ;
while ( ( urlFileStack ! = null ) & & ( count > 0 ) & & ( urlFileStack . size ( ) > 0 ) ) {
// flush some entries from disc to ram stack
try {
// one from the top:
2009-01-30 16:33:00 +01:00
Row . Entry t = urlFileStack . pop ( ) ;
2008-09-17 23:45:21 +02:00
if ( t = = null ) break ;
pushHashToDomainStacks ( new String ( t . getColBytes ( 0 ) ) , false ) ;
count - - ;
if ( urlFileStack . size ( ) = = 0 ) break ;
// one from the bottom:
t = urlFileStack . pot ( ) ;
if ( t = = null ) break ;
pushHashToDomainStacks ( new String ( t . getColBytes ( 0 ) ) , false ) ;
count - - ;
} catch ( final IOException e ) {
break ;
}
}
}
private void shiftFileToRAM ( final int wantedsize ) {
while ( ( urlFileStack ! = null ) & & ( urlRAMStack . size ( ) < = wantedsize ) & & ( urlFileStack . size ( ) > 0 ) ) {
// flush some entries from disc to ram stack
try {
// one from the top:
2009-01-30 16:33:00 +01:00
Row . Entry t = urlFileStack . pop ( ) ;
2008-09-17 23:45:21 +02:00
if ( t = = null ) break ;
urlRAMStack . add ( new String ( t . getColBytes ( 0 ) ) ) ;
if ( urlFileStack . size ( ) = = 0 ) break ;
// one from the bottom:
t = urlFileStack . pot ( ) ;
if ( t = = null ) break ;
urlRAMStack . add ( new String ( t . getColBytes ( 0 ) ) ) ;
} catch ( final IOException e ) {
break ;
}
}
}
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 ( ) ) ) {
2008-10-03 20:57:02 +02:00
//serverLog.logWarning("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
2008-09-17 23:45:21 +02:00
pushHashToDomainStacks ( entry . url ( ) . hash ( ) , true ) ;
2008-07-03 15:08:37 +02:00
}
2008-09-17 23:45:21 +02:00
private void pushHashToDomainStacks ( final String hash , boolean flush ) {
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-09-17 23:45:21 +02:00
if ( flush & & ( domainStacks . size ( ) > 100 ) | | ( sizeDomainStacks ( ) > 1000 ) ) {
2008-11-22 12:25:01 +01:00
flushOnceDomStacks ( urlRAMStack . size ( ) < 100 , true , 100 ) ; // when the ram stack is small, flush it there
2005-09-25 03:09:21 +02:00
}
}
2008-10-08 20:26:36 +02:00
/ * *
* 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
* @param profile
* @return a url in a CrawlEntry object
* @throws IOException
* /
public synchronized CrawlEntry pop ( boolean delay , CrawlProfile profile ) throws IOException {
2008-09-17 23:45:21 +02:00
// returns a crawl entry 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-09-17 23:45:21 +02:00
//result = urlRAMStack.remove(0);
Iterator < String > i = urlRAMStack . iterator ( ) ;
String urlhash ;
long waitingtime , min = Long . MAX_VALUE ;
String besthash = null ;
while ( i . hasNext ( ) ) {
urlhash = i . next ( ) ;
waitingtime = CrawlEntry . waitingRemainingGuessed ( urlhash , minimumLocalDelta , minimumGlobalDelta ) ;
if ( waitingtime = = 0 ) {
// zero waiting is a good one
result = urlhash ;
i . remove ( ) ;
min = Long . MAX_VALUE ; // that causes that the if at the end of this loop is not used
besthash = null ;
break ;
}
if ( waitingtime < min ) {
min = waitingtime ;
besthash = urlhash ;
}
}
if ( min < = 500 & & besthash ! = null ) {
// find that entry that was best end remove it
i = urlRAMStack . iterator ( ) ;
while ( i . hasNext ( ) ) {
urlhash = i . next ( ) ;
if ( urlhash . equals ( besthash ) ) {
// zero waiting is a good one
result = urlhash ;
i . remove ( ) ;
break ;
}
}
}
2007-02-21 17:23:31 +01:00
}
2008-09-17 23:45:21 +02:00
// the next options use the domain stack. If this is not filled enough, they dont work at all
// so just fill them up with some stuff
if ( result = = null ) shiftFileToDomStacks ( 1000 ) ;
2008-09-18 00:13:25 +02:00
// 2nd-b: check domainStacks for best match between stack size and retrieval time
String maxhash = null ;
if ( ( result = = null ) & & ( domainStacks . size ( ) > 0 ) ) synchronized ( domainStacks ) {
// we order all domains by the number of entries per domain
// then we iterate through these domains in descending entry order
// and take that one, that has a zero waiting time
final Iterator < Map . Entry < String , LinkedList < String > > > i = domainStacks . entrySet ( ) . iterator ( ) ;
Map . Entry < String , LinkedList < String > > entry ;
String domhash ;
LinkedList < String > domlist ;
final TreeMap < Integer , String > hitlist = new TreeMap < Integer , String > ( ) ;
int count = 0 ;
// first collect information about sizes of the domain lists
while ( i . hasNext ( ) ) {
entry = i . next ( ) ;
domhash = entry . getKey ( ) ;
domlist = entry . getValue ( ) ;
hitlist . put ( Integer . valueOf ( domlist . size ( ) * 100 + count + + ) , domhash ) ;
}
// now iterate in descending order and fetch that one,
// that is acceptable by the minimumDelta constraint
long waitingtime ;
while ( hitlist . size ( ) > 0 ) {
domhash = hitlist . remove ( hitlist . lastKey ( ) ) ;
if ( maxhash = = null ) maxhash = domhash ; // remember first entry
waitingtime = CrawlEntry . waitingRemainingGuessed ( domhash , minimumLocalDelta , minimumGlobalDelta ) ;
if ( waitingtime < 100 ) {
domlist = domainStacks . get ( domhash ) ;
result = domlist . removeFirst ( ) ;
if ( domlist . size ( ) = = 0 ) domainStacks . remove ( domhash ) ;
break ;
}
}
}
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
2008-09-17 23:45:21 +02:00
// 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 ;
2008-09-17 23:45:21 +02:00
long waitingtime , min = Long . MAX_VALUE ;
String besthash = 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 ( ) ;
2008-09-17 23:45:21 +02:00
waitingtime = CrawlEntry . waitingRemainingGuessed ( domhash , minimumLocalDelta , minimumGlobalDelta ) ;
if ( waitingtime = = 0 ) {
// zero waiting is a good one
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 ( ) ;
2008-09-17 23:45:21 +02:00
min = Long . MAX_VALUE ; // that causes that the if at the end of this loop is not used
besthash = null ;
2007-02-21 17:23:31 +01:00
break ;
}
2008-09-17 23:45:21 +02:00
if ( waitingtime < min ) {
min = waitingtime ;
besthash = domhash ;
2007-02-21 17:23:31 +01:00
}
}
2008-09-17 23:45:21 +02:00
if ( min < = 500 & & besthash ! = null ) {
domlist = domainStacks . get ( besthash ) ;
2008-06-06 18:01:27 +02:00
result = domlist . removeFirst ( ) ;
2008-09-17 23:45:21 +02:00
if ( domlist . size ( ) = = 0 ) domainStacks . remove ( besthash ) ;
2007-02-21 17:23:31 +01:00
}
}
2008-09-18 00:13:25 +02:00
// 2nd-c: if we did yet not choose any entry, we simply take that one with the most entries
if ( ( result = = null ) & & ( maxhash ! = null ) ) {
LinkedList < String > domlist = domainStacks . get ( maxhash ) ;
if ( domlist ! = null ) {
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 ) ) {
2009-01-30 16:33:00 +01:00
final Row . 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.
2009-03-13 15:56:25 +01:00
urlFileStack . clear ( ) ;
2009-01-31 00:33:47 +01:00
Log . logSevere ( " BALANCER " , " get() failed to fetch entry from file stack. reset stack file. " ) ;
2007-03-07 10:08:13 +01:00
} 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
2008-09-17 23:45:21 +02:00
long waitingtime = CrawlEntry . waitingRemainingGuessed ( nexthash , minimumLocalDelta , minimumGlobalDelta ) ;
if ( waitingtime = = 0 ) {
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
}
}
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 ) {
2009-01-31 00:33:47 +01:00
Log . logSevere ( " 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 int s = urlFileIndex . size ( ) ;
2009-01-30 16:33:00 +01:00
Row . Entry rowEntry = urlFileIndex . remove ( result . getBytes ( ) ) ;
2007-06-06 09:53:56 +02:00
if ( rowEntry = = null ) {
2008-09-17 23:45:21 +02:00
throw new IOException ( " 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 ( ) ) ;
2007-06-06 09:53:56 +02:00
}
2008-08-02 15:57:00 +02:00
assert urlFileIndex . size ( ) + 1 = = s : " urlFileIndex.size() = " + urlFileIndex . size ( ) + " , s = " + s + " , result = " + result ;
2008-10-08 20:26:36 +02:00
final CrawlEntry crawlEntry = new CrawlEntry ( rowEntry ) ;
// at this point we must check if the crawlEntry has relevancy because the crawl profile still exists
// if not: return null. A calling method must handle the null value and try again
if ( profile ! = null & & ! profile . hasEntry ( crawlEntry . profileHandle ( ) ) ) return null ;
long sleeptime = crawlEntry . waitingRemaining ( minimumLocalDelta , minimumGlobalDelta ) ; // this uses the robots.txt database and may cause a loading of robots.txt from the server
2008-09-17 23:45:21 +02:00
if ( delay & & sleeptime > 0 ) {
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
2008-09-08 23:50:37 +02:00
// this is only to protection against the worst case, where the crawler could
2007-02-21 17:23:31 +01:00
// behave in a DoS-manner
2009-01-31 00:33:47 +01:00
Log . logInfo ( " BALANCER " , " forcing crawl-delay of " + sleeptime + " milliseconds for " + crawlEntry . url ( ) . getHost ( ) + ( ( sleeptime > Math . max ( minimumLocalDelta , minimumGlobalDelta ) ) ? " (caused by robots.txt) " : " " ) ) ;
2008-08-02 14:12:04 +02:00
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-09-17 23:45:21 +02:00
crawlEntry . updateAccess ( ) ;
2007-06-06 09:53:56 +02:00
return crawlEntry ;
2007-02-21 17:23:31 +01:00
}
2008-07-03 15:08:37 +02:00
2008-09-17 23:45:21 +02:00
/ * *
* return top - elements from the crawl stack
* we do not produce here more entries than exist on the stack
* because otherwise the balancing does not work properly
* @param count
* @return
* @throws IOException
* /
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-09-17 23:45:21 +02:00
// ensure that the domain stacks are filled enough
shiftFileToDomStacks ( count ) ;
// flush from the domain stacks first until they are empty
if ( ( domainStacksNotEmpty ( ) ) & & ( urlRAMStack . size ( ) < = count ) ) {
2008-11-22 12:25:01 +01:00
flushOnceDomStacks ( true , true , 100 ) ;
2008-09-17 23:45:21 +02:00
}
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
2008-11-22 12:25:01 +01:00
flushOnceDomStacks ( true , false , 100 ) ;
2005-09-25 03:09:21 +02:00
}
2008-07-03 15:08:37 +02:00
2008-09-17 23:45:21 +02:00
// if the ram is still not full enough, use the file stack
shiftFileToRAM ( count ) ;
// finally, construct a list using the urlRAMStack which was filled with this procedure
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 ) ;
2009-01-30 16:33:00 +01:00
final Row . 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
2009-01-30 16:33:00 +01:00
private Iterator < Row . Entry > rowIterator ;
2007-03-16 14:25:56 +01:00
public EntryIterator ( ) throws IOException {
2009-02-24 11:40:20 +01:00
rowIterator = urlFileIndex . rows ( ) ;
2007-03-16 14:25:56 +01:00
}
public boolean hasNext ( ) {
return ( rowIterator = = null ) ? false : rowIterator . hasNext ( ) ;
}
2008-05-06 02:32:41 +02:00
public CrawlEntry next ( ) {
2009-01-30 16:33:00 +01:00
final Row . 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
}
}