boxtec
/
yacy_search_server
mirror of https://github.com/yacy/yacy_search_server
4
1
Fork 0
Code Issues Projects Releases Wiki Activity

removed strange retry logic from DHT transfer

Browse Source 
git-svn-id: https://svn.berlios.de/svnroot/repos/yacy/trunk@5564 6c8d7289-2bf4-0310-a012-ef5d649a1542
pull/1/head
orbiter 16 years ago
parent 12ec5132ce
commit d3e33fd6c1
3 changed files with 42 additions and 141 deletions
Show Stats Download Patch File Download Diff File

5
source/de/anomic/plasma/plasmaDHTFlush.java
Unescape View File

@ -216,10 +216,9 @@ public class plasmaDHTFlush extends Thread {
oldDHTChunk = null;
}
// handover chunk to transfer worker
// hand-over chunk to transfer worker
if ((newDHTChunk.containerSize() > 0) || (newDHTChunk.getStatus() == plasmaDHTChunk.chunkStatus_FILLED)) {
this.worker = new plasmaDHTTransfer(this.log, this.wordIndex.seedDB, this.wordIndex.peerActions, this.seed, newDHTChunk, this.gzipBody4Transfer, this.timeout4Transfer, 5);
this.worker.setTransferMode(plasmaDHTTransfer.TRANSFER_MODE_FLUSH);
this.worker = new plasmaDHTTransfer(this.log, this.wordIndex.seedDB, this.wordIndex.peerActions, this.seed, newDHTChunk, this.gzipBody4Transfer, this.timeout4Transfer);
this.worker.start();
}
}

173
source/de/anomic/plasma/plasmaDHTTransfer.java
Unescape View File

@ -56,8 +56,6 @@ public class plasmaDHTTransfer extends Thread {
// other fields
private final yacySeedDB seedDB;
private final yacyPeerActions peerActions;
private final int maxRetry;
private int transferMode = TRANSFER_MODE_DISTRIBUTION;
Log log;
public plasmaDHTTransfer(
@ -67,8 +65,7 @@ public class plasmaDHTTransfer extends Thread {
final yacySeed destSeed,
final plasmaDHTChunk dhtChunk,
final boolean gzipBody,
final int timeout,
final int retries
final int timeout
) {
super(new ThreadGroup("TransferIndexThreadGroup"), "TransferIndexWorker_" + destSeed.getName());
this.log = log;
@ -77,13 +74,8 @@ public class plasmaDHTTransfer extends Thread {
this.gzipBody4Transfer = gzipBody;
this.timeout4Transfer = timeout;
this.dhtChunk = dhtChunk;
this.maxRetry = retries;
this.seed = destSeed;
}
public void setTransferMode(final int mode) {
this.transferMode = mode;
}
public void run() {
try {
@ -133,138 +125,49 @@ public class plasmaDHTTransfer extends Thread {
* - the retry counter limit was exceeded
*/
this.transferStatus = plasmaDHTChunk.chunkStatus_RUNNING;
long retryCount = 0;
final long start = System.currentTimeMillis();
while (true) {
// testing if we were aborted
if (this.isAborted()) return;
// transfering seleted words to remote peer
this.transferStatusMessage = "Running: Transfering chunk to target " + this.seed.hash + "/" + this.seed.getName();
final HashMap<String, Object> result = yacyClient.transferIndex(this.seedDB, this.seed, this.dhtChunk.containers(), this.dhtChunk.urlCacheMap(), this.gzipBody4Transfer, this.timeout4Transfer);
final String error = (String) result.get("result");
if (error == null) {
// words successfully transfered
this.transferTime = System.currentTimeMillis() - start;
this.payloadSize = ((Integer)result.get("payloadSize")).intValue();
this.log.logInfo("Index transfer of " + this.dhtChunk.indexCount() +
" entries " + this.dhtChunk.containerSize() +
" words [" + this.dhtChunk.firstContainer().getWordHash() + " .. " + this.dhtChunk.lastContainer().getWordHash() + "]" +
" and " + this.dhtChunk.urlCacheMap().size() + " URLs" +
" to peer " + this.seed.getName() + ":" + this.seed.hash +
" in " + (this.transferTime / 1000) +
" seconds successful (" + (1000 * this.dhtChunk.indexCount() / (this.transferTime + 1)) +
" words/s, " + this.payloadSize + " Bytes)");
// if the peer has set a pause time and we are in flush mode (index transfer)
// then we pause for a while now
if (this.transferMode == TRANSFER_MODE_FLUSH) {
final long pause = getBusyTime(result);
if (pause != -1) {
this.transferStatusMessage = "Finished: Transfer of chunk to target " + this.seed.hash + "/" + this.seed.getName() + ". Pausing " + pause + " ms.";
this.pause(pause);
}
} else {
this.transferStatusMessage = "Finished: Transfer of chunk to target " + this.seed.hash + "/" + this.seed.getName();
}
// transfer of chunk finished
this.transferStatus = plasmaDHTChunk.chunkStatus_COMPLETE;
retryCount = 0;
break;
}
// testing if we were aborted
if (this.isAborted()) return;
// transferring selected words to remote peer
this.transferStatusMessage = "Running: Transfering chunk to target " + this.seed.hash + "/" + this.seed.getName();
final HashMap<String, Object> result = yacyClient.transferIndex(this.seedDB, this.seed, this.dhtChunk.containers(), this.dhtChunk.urlCacheMap(), this.gzipBody4Transfer, this.timeout4Transfer);
final String error = (String) result.get("result");
if (error == null) {
// words successfully transfered
this.transferTime = System.currentTimeMillis() - start;
this.payloadSize = ((Integer)result.get("payloadSize")).intValue();
// inc retry counter
retryCount++;
if (this.isAborted()) return;
boolean reconnectNeeded = false;
long pauseTime = 1;
this.log.logInfo("Index transfer of " + this.dhtChunk.indexCount() +
" entries " + this.dhtChunk.containerSize() +
" words [" + this.dhtChunk.firstContainer().getWordHash() + " .. " + this.dhtChunk.lastContainer().getWordHash() + "]" +
" and " + this.dhtChunk.urlCacheMap().size() + " URLs" +
" to peer " + this.seed.getName() + ":" + this.seed.hash +
" in " + (this.transferTime / 1000) +
" seconds successful (" + (1000 * this.dhtChunk.indexCount() / (this.transferTime + 1)) +
" words/s, " + this.payloadSize + " Bytes)");
if (error.equals("busy")) {
// get pause time that was requested by the remote peer
pauseTime = getBusyTime(result);
if (pauseTime == -1) pauseTime = 60000;
this.transferStatusMessage = "Peer " + this.seed.getName() + ":" + this.seed.hash + " is busy. Waiting " + pauseTime + " ms.";
this.log.logInfo(this.transferStatusMessage);
} else {
this.transferStatusMessage = "Transfer to peer " + this.seed.getName() + ":" + this.seed.hash + " failed:'" + error + "', Trying to reconnect ...";
// force disconnection of peer
peerActions.peerDeparture(this.seed, "DHT Transfer: " + this.transferStatusMessage);
this.log.logWarning(this.transferStatusMessage);
// calculate pause time
pauseTime = retryCount * 10000;
reconnectNeeded = true;
}
// if the peer has set a pause time and we are in flush mode (index transfer)
// then we pause for a while now
this.transferStatusMessage = "Finished: Transfer of chunk to target " + this.seed.hash + "/" + this.seed.getName();
// if the retry counter limit was not exceeded we'll retry it in a few seconds
if (retryCount > this.maxRetry) {
this.transferStatusMessage = "Transfer aborted. Retry limit reached.";
this.transferStatus = plasmaDHTChunk.chunkStatus_FAILED;
return;
}
// sleep for a while
this.pause(pauseTime);
// transfer of chunk finished
this.transferStatus = plasmaDHTChunk.chunkStatus_COMPLETE;
return;
}
if (this.isAborted()) return;
if (error.equals("busy")) {
this.transferStatusMessage = "Peer " + this.seed.getName() + ":" + this.seed.hash + " is busy.";
this.log.logInfo(this.transferStatusMessage);
} else {
this.transferStatusMessage = "Transfer to peer " + this.seed.getName() + ":" + this.seed.hash + " failed:'" + error + "', Trying to reconnect ...";
// reconnect to peer if needed
if (reconnectNeeded) {
/* loop until
* - we have successfully done a peer ping or
* - the retry counter limit was exceeded
*/
while (true) {
// testing if we were aborted ...
if (this.isAborted())
return;
// force disconnection of peer
peerActions.peerDeparture(this.seed, "DHT Transfer: " + this.transferStatusMessage);
this.log.logWarning(this.transferStatusMessage);
// doing a peer ping to the remote seed
final int added = yacyClient.publishMySeed(this.seedDB.mySeed(), this.peerActions, this.seed.getPublicAddress(), this.seed.hash);
if (added < 0) {
// inc. retry counter
retryCount++;
this.transferStatusMessage = "Disconnected peer: Peer ping failed. " + ((retryCount > 5) ? "Transfer aborted." : "Retry " + retryCount);
if (retryCount > this.maxRetry) return;
this.pause(retryCount * 10000);
continue;
}
this.seedDB.getConnected(this.seed.hash);
this.transferStatusMessage = "running";
break;
}
}
}
}
@SuppressWarnings("unchecked")
private long getBusyTime(final HashMap<String, Object> result) {
int pause = -1;
final Object transferRWIResult = result.get("resultTransferRWI");
assert transferRWIResult instanceof HashMap;
if (transferRWIResult != null && ((HashMap<String, String>) transferRWIResult).containsKey("pause")) {
final String pauseStr = ((HashMap<String, String>) transferRWIResult).get("pause");
try { pause = Integer.valueOf(pauseStr).intValue(); } catch (final NumberFormatException numEx){}
if (pause < 0) pause = 5000;
else if (pause > 30000) pause = 30000;
}
return pause;
}
private void pause(final long sleepTime) throws InterruptedException {
if (sleepTime == 0) return;
long sleepCounter = sleepTime / 1000;
final long sleepRest = sleepTime % 1000;
while (!this.isAborted() && sleepCounter > 0) {
sleepCounter--;
Thread.sleep(1000);
}
if (sleepRest > 0) Thread.sleep(sleepRest);
}
}

5
source/de/anomic/plasma/plasmaSwitchboard.java
Unescape View File

@ -2015,8 +2015,7 @@ public final class plasmaSwitchboard extends serverAbstractSwitch<IndexingStack.
// getting distribution configuration values
final boolean gzipBody = getConfig(plasmaSwitchboardConstants.INDEX_DIST_GZIP_BODY, "false").equalsIgnoreCase("true");
final int timeout = (int)getConfigLong(plasmaSwitchboardConstants.INDEX_DIST_TIMEOUT, 60000);
final int retries = 0;
// starting up multiple DHT transfer threads
yacySeed seed;
long firstdist, lastdist;
@ -2034,7 +2033,7 @@ public final class plasmaSwitchboard extends serverAbstractSwitch<IndexingStack.
firstdist = yacySeed.dhtDistance(dhtChunk.firstContainer().getWordHash(), seed);
lastdist = yacySeed.dhtDistance(dhtChunk.lastContainer().getWordHash(), seed);
if (log != null) log.logInfo("Selected DHT target peer " + seed.getName() + ":" + seed.hash + ", distance2first = " + firstdist + ", distance2last = " + lastdist);
final plasmaDHTTransfer t = new plasmaDHTTransfer(log, webIndex.seedDB, webIndex.peerActions, seed, dhtChunk, gzipBody, timeout, retries);
final plasmaDHTTransfer t = new plasmaDHTTransfer(log, webIndex.seedDB, webIndex.peerActions, seed, dhtChunk, gzipBody, timeout);
t.start();
transfer.add(t);
} else {

Write Preview
Loading…
Cancel
Save