2006-02-20 00:47:45 +01:00
// plasmaDHTFlush.java
// ------------------------------
// part of YaCy
// (C) by Michael Peter Christen; mc@anomic.de
// first published on http://www.anomic.de
// Frankfurt, Germany, 2005, 2006
//
// This Class was written by Martin Thelian
//
// 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 ;
import de.anomic.server.logging.serverLog ;
import de.anomic.yacy.yacySeed ;
public class plasmaDHTFlush extends Thread {
private yacySeed seed = null ;
private boolean delete = false ;
private boolean finished = false ;
private boolean gzipBody4Transfer = false ;
private int timeout4Transfer = 60000 ;
private int transferedEntryCount = 0 ;
2006-06-14 11:40:42 +02:00
private long transferedBytes = 0 ;
2006-02-20 00:47:45 +01:00
private int transferedContainerCount = 0 ;
private String status = " Running " ;
2007-01-01 13:27:05 +01:00
private String oldStartingPointHash = " AAAAAAAAAAAA " , startPointHash = " AAAAAAAAAAAA " ;
2006-02-20 00:47:45 +01:00
private int initialWordsDBSize = 0 ;
private int chunkSize = 500 ;
private final long startingTime = System . currentTimeMillis ( ) ;
private final plasmaSwitchboard sb ;
private plasmaDHTTransfer worker = null ;
private serverLog log ;
private plasmaWordIndex wordIndex ;
public plasmaDHTFlush ( serverLog log , plasmaWordIndex wordIndex , yacySeed seed , boolean delete , boolean gzipBody , int timeout ) {
super ( new ThreadGroup ( " TransferIndexThreadGroup " ) , " TransferIndex_ " + seed . getName ( ) ) ;
this . log = log ;
this . wordIndex = wordIndex ;
this . seed = seed ;
this . delete = delete ;
this . sb = plasmaSwitchboard . getSwitchboard ( ) ;
2006-05-13 17:28:57 +02:00
this . initialWordsDBSize = this . sb . wordIndex . size ( ) ;
2006-02-20 00:47:45 +01:00
this . gzipBody4Transfer = gzipBody ;
this . timeout4Transfer = timeout ;
//this.maxOpenFiles4Transfer = (int) sb.getConfigLong("indexTransfer.maxOpenFiles",800);
}
public void run ( ) {
2006-05-13 17:28:57 +02:00
this . performTransferWholeIndex ( ) ;
2006-02-20 00:47:45 +01:00
}
public void stopIt ( boolean wait ) throws InterruptedException {
this . finished = true ;
2006-06-14 11:40:42 +02:00
if ( this . worker ! = null ) this . worker . stopIt ( ) ;
2006-02-20 00:47:45 +01:00
if ( wait ) this . join ( ) ;
}
public boolean isFinished ( ) {
return this . finished ;
}
public boolean deleteIndex ( ) {
return this . delete ;
}
public int [ ] getIndexCount ( ) {
plasmaDHTTransfer workerThread = this . worker ;
if ( workerThread ! = null ) {
return new int [ ] { this . chunkSize , workerThread . dhtChunk . indexCount ( ) } ;
}
return new int [ ] { this . chunkSize , 500 } ;
}
public int getTransferedEntryCount ( ) {
return this . transferedEntryCount ;
}
public int getTransferedContainerCount ( ) {
return this . transferedContainerCount ;
}
2006-06-14 11:40:42 +02:00
public long getTransferedBytes ( ) {
return this . transferedBytes ;
}
2006-02-20 00:47:45 +01:00
public float getTransferedContainerPercent ( ) {
2006-05-13 17:28:57 +02:00
long currentWordsDBSize = this . sb . wordIndex . size ( ) ;
if ( this . initialWordsDBSize = = 0 ) return 100 ;
else if ( currentWordsDBSize > = this . initialWordsDBSize ) return 0 ;
2006-02-20 00:47:45 +01:00
//else return (float) ((initialWordsDBSize-currentWordsDBSize)/(initialWordsDBSize/100));
2006-05-13 17:28:57 +02:00
else return ( this . transferedContainerCount * 100 / this . initialWordsDBSize ) ;
2006-02-20 00:47:45 +01:00
}
2006-07-19 15:52:33 +02:00
public int getTransferedEntrySpeed ( ) {
2006-05-13 17:28:57 +02:00
long transferTime = System . currentTimeMillis ( ) - this . startingTime ;
2006-02-20 00:47:45 +01:00
if ( transferTime < = 0 ) transferTime = 1 ;
2006-05-13 17:28:57 +02:00
return ( int ) ( ( 1000 * this . transferedEntryCount ) / transferTime ) ;
2006-02-20 00:47:45 +01:00
}
public yacySeed getSeed ( ) {
return this . seed ;
}
public String [ ] getStatus ( ) {
plasmaDHTTransfer workerThread = this . worker ;
if ( workerThread ! = null ) {
2006-02-22 00:08:07 +01:00
return new String [ ] { this . status , workerThread . getStatusMessage ( ) } ;
2006-02-20 00:47:45 +01:00
}
return new String [ ] { this . status , " Not running " } ;
}
public String [ ] getRange ( ) {
plasmaDHTTransfer workerThread = this . worker ;
if ( workerThread ! = null ) {
2006-05-13 17:28:57 +02:00
return new String [ ] { " [ " + this . oldStartingPointHash + " .. " + this . startPointHash + " ] " ,
2006-07-04 16:47:27 +02:00
" [ " + workerThread . dhtChunk . firstContainer ( ) . getWordHash ( ) + " .. " + workerThread . dhtChunk . lastContainer ( ) . getWordHash ( ) + " ] " } ;
2006-02-20 00:47:45 +01:00
}
2006-05-13 17:28:57 +02:00
return new String [ ] { " [ " + this . oldStartingPointHash + " .. " + this . startPointHash + " ] " , " [------------..------------] " } ;
2006-02-20 00:47:45 +01:00
}
public void performTransferWholeIndex ( ) {
plasmaDHTChunk newDHTChunk = null , oldDHTChunk = null ;
try {
// initial startingpoint of intex transfer is "------------"
2006-05-13 17:28:57 +02:00
this . log . logFine ( " Selected hash " + this . startPointHash + " as start point for index distribution of whole index " ) ;
2006-02-20 00:47:45 +01:00
/ * Loop until we have
* - finished transfer of whole index
* - detected a server shutdown or user interruption
* - detected a failure
* /
2006-05-13 17:28:57 +02:00
long iteration = 0 ;
2006-02-20 00:47:45 +01:00
2006-05-13 17:28:57 +02:00
while ( ! this . finished & & ! Thread . currentThread ( ) . isInterrupted ( ) ) {
2006-02-20 00:47:45 +01:00
iteration + + ;
oldDHTChunk = newDHTChunk ;
// selecting 500 words to transfer
this . status = " Running: Selecting chunk " + iteration ;
2006-12-06 13:51:46 +01:00
newDHTChunk = new plasmaDHTChunk ( this . log , this . wordIndex , this . chunkSize / 3 * 2 , this . chunkSize , - 1 , this . startPointHash ) ;
2006-02-20 00:47:45 +01:00
/ * If we havn ' t selected a word chunk this could be because of
* a ) no words are left in the index
* b ) max open file limit was exceeded
* /
2006-05-13 17:28:57 +02:00
if ( nothingSelected ( newDHTChunk ) ) {
2006-05-14 15:27:50 +02:00
if ( this . sb . wordIndex . size ( ) > 0 & & this . delete ) {
2006-02-20 00:47:45 +01:00
// if there are still words in the index we try it again now
2006-05-13 17:28:57 +02:00
this . startPointHash = " ------------ " ;
2006-02-20 00:47:45 +01:00
} else {
// otherwise we could end transfer now
2006-05-13 17:28:57 +02:00
this . log . logFine ( " No index available for index transfer, hash start-point " + this . startPointHash ) ;
2006-02-20 00:47:45 +01:00
this . status = " Finished. " + iteration + " chunks transfered. " ;
2006-05-13 17:28:57 +02:00
this . finished = true ;
2006-02-20 00:47:45 +01:00
}
} else {
// getting start point for next DHT-selection
2006-05-13 17:28:57 +02:00
this . oldStartingPointHash = this . startPointHash ;
2006-07-04 16:47:27 +02:00
this . startPointHash = newDHTChunk . lastContainer ( ) . getWordHash ( ) ; // DHT targets must have greater hashes
2006-02-20 00:47:45 +01:00
2006-07-04 16:47:27 +02:00
this . log . logInfo ( " Index selection of " + newDHTChunk . indexCount ( ) + " words [ " + newDHTChunk . firstContainer ( ) . getWordHash ( ) + " .. " + newDHTChunk . lastContainer ( ) . getWordHash ( ) + " ] " +
2006-02-20 00:47:45 +01:00
" in " +
2006-05-13 17:28:57 +02:00
( newDHTChunk . getSelectionTime ( ) / 1000 ) + " seconds ( " +
( 1000 * newDHTChunk . indexCount ( ) / ( newDHTChunk . getSelectionTime ( ) + 1 ) ) + " words/s) " ) ;
2006-02-20 00:47:45 +01:00
}
// query status of old worker thread
2006-05-13 17:28:57 +02:00
if ( this . worker ! = null ) {
2006-02-20 00:47:45 +01:00
this . status = " Finished: Selecting chunk " + iteration ;
2006-05-13 17:28:57 +02:00
this . worker . join ( ) ;
if ( this . worker . getStatus ( ) ! = plasmaDHTChunk . chunkStatus_COMPLETE ) {
2006-02-20 00:47:45 +01:00
// if the transfer failed we abort index transfer now
2006-05-13 17:28:57 +02:00
this . status = " Aborted because of Transfer error: \ n " + this . worker . dhtChunk . getStatus ( ) ;
2006-02-20 00:47:45 +01:00
// abort index transfer
return ;
}
2006-05-13 17:28:57 +02:00
// calculationg the new transfer size
this . calculateNewChunkSize ( ) ;
// counting transfered containers / entries
this . transferedEntryCount + = oldDHTChunk . indexCount ( ) ;
this . transferedContainerCount + = oldDHTChunk . containerSize ( ) ;
2006-06-14 11:40:42 +02:00
this . transferedBytes + = this . worker . getPayloadSize ( ) ;
this . worker = null ;
2006-05-13 17:28:57 +02:00
// deleting transfered words from index
if ( this . delete ) {
this . status = " Running: Deleting chunk " + iteration ;
2006-10-13 01:14:41 +02:00
String urlReferences = oldDHTChunk . deleteTransferIndexes ( ) ;
2006-05-13 17:28:57 +02:00
this . log . logFine ( " Deleted from " + oldDHTChunk . containerSize ( ) + " transferred RWIs locally " + urlReferences + " URL references " ) ;
}
oldDHTChunk = null ;
2006-02-20 00:47:45 +01:00
}
// handover chunk to transfer worker
2006-05-13 17:28:57 +02:00
if ( ( newDHTChunk . containerSize ( ) > 0 ) | | ( newDHTChunk . getStatus ( ) = = plasmaDHTChunk . chunkStatus_FILLED ) ) {
this . worker = new plasmaDHTTransfer ( this . log , this . seed , newDHTChunk , this . gzipBody4Transfer , this . timeout4Transfer , 5 ) ;
2006-06-14 11:40:42 +02:00
this . worker . setTransferMode ( plasmaDHTTransfer . TRANSFER_MODE_FLUSH ) ;
2006-05-13 17:28:57 +02:00
this . worker . start ( ) ;
2006-02-20 00:47:45 +01:00
}
}
// if we reach this point we were aborted by the user or by server shutdown
2006-05-13 17:28:57 +02:00
if ( this . sb . wordIndex . size ( ) > 0 ) this . status = " aborted " ;
2006-02-20 00:47:45 +01:00
} catch ( Exception e ) {
this . status = " Error: " + e . getMessage ( ) ;
2006-05-13 17:28:57 +02:00
this . log . logWarning ( " Index transfer to peer " + this . seed . getName ( ) + " : " + this . seed . hash + " failed:' " + e . getMessage ( ) + " ' " , e ) ;
2006-02-20 00:47:45 +01:00
} finally {
2006-05-13 17:28:57 +02:00
if ( this . worker ! = null ) {
this . worker . stopIt ( ) ;
try { this . worker . join ( ) ; } catch ( Exception e ) { }
2006-02-20 00:47:45 +01:00
}
}
}
2006-05-13 17:28:57 +02:00
private void calculateNewChunkSize ( ) {
// getting the transfered chunk size
this . chunkSize = this . worker . dhtChunk . indexCount ( ) ;
// getting the chunk selection time
long selectionTime = this . worker . dhtChunk . getSelectionTime ( ) ;
// getting the chunk transfer time
long transferTime = this . worker . getTransferTime ( ) ;
// calculationg the new chunk size
if ( transferTime > 60 * 1000 & & this . chunkSize > 200 ) {
this . chunkSize - = 100 ;
} else if ( selectionTime < transferTime ) {
this . chunkSize + = 100 ;
} else if ( selectionTime > = selectionTime & & this . chunkSize > 200 ) {
this . chunkSize - = 100 ;
}
}
private static boolean nothingSelected ( plasmaDHTChunk newDHTChunk ) {
return ( newDHTChunk = = null ) | |
( newDHTChunk . containerSize ( ) = = 0 ) | |
( newDHTChunk . getStatus ( ) = = plasmaDHTChunk . chunkStatus_FAILED ) ;
}
2006-02-20 00:47:45 +01:00
}