Author: mrogers
Date: 2006-11-18 12:53:43 +0000 (Sat, 18 Nov 2006)
New Revision: 10983
Modified:
trunk/apps/load-balancing-sims/phase7/sim/Node.java
trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkInsertHandler.java
trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkRequestHandler.java
trunk/apps/load-balancing-sims/phase7/sim/handlers/MessageHandler.java
trunk/apps/load-balancing-sims/phase7/sim/handlers/SskInsertHandler.java
trunk/apps/load-balancing-sims/phase7/sim/handlers/SskRequestHandler.java
Log:
CHK inserts were occasionally being logged as both successes and failures
Modified: trunk/apps/load-balancing-sims/phase7/sim/Node.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/sim/Node.java 2006-11-18 09:11:16 UTC
(rev 10982)
+++ trunk/apps/load-balancing-sims/phase7/sim/Node.java 2006-11-18 12:53:43 UTC
(rev 10983)
@@ -420,13 +420,12 @@
ih.start();
}
- public void searchSucceeded (MessageHandler m)
+ public void increaseSearchRate()
{
- log (m + " succeeded");
if (USE_THROTTLE) searchThrottle.increaseRate();
}
- public void reduceSearchRate (MessageHandler m)
+ public void decreaseSearchRate()
{
if (USE_THROTTLE) searchThrottle.decreaseRate();
}
Modified:
trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkInsertHandler.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkInsertHandler.java
2006-11-18 09:11:16 UTC (rev 10982)
+++ trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkInsertHandler.java
2006-11-18 12:53:43 UTC (rev 10983)
@@ -11,6 +11,7 @@
private HashSet<Peer> receivers; // Peers that should receive data
private Block[] blocks; // Store incoming blocks for forwarding
private int blocksReceived = 0;
+ private boolean transferred = false; // Did we send the data to anyone?
public ChkInsertHandler (ChkInsert i, Node node, Peer prev)
{
@@ -40,7 +41,7 @@
else if (m instanceof RejectedLoop)
handleRejectedLoop ((RejectedLoop) m);
else if (m instanceof RejectedOverload)
- handleOverload ((RejectedOverload) m, src);
+ handleRejectedOverload ((RejectedOverload) m);
else if (m instanceof RouteNotFound)
handleRouteNotFound ((RouteNotFound) m);
else if (m instanceof InsertReply)
@@ -90,6 +91,7 @@
private void handleCompleted (TransfersCompleted tc, Peer src)
{
receivers.remove (src);
+ transferred = true;
considerFinishing();
}
@@ -113,14 +115,14 @@
private void handleInsertReply (InsertReply ir)
{
if (searchState != ACCEPTED) node.log (ir + " out of order");
- if (prev == null) node.searchSucceeded (this);
+ if (prev == null) node.increaseSearchRate();
else prev.sendMessage (ir); // Forward the message
finish();
}
protected void sendReply()
{
- if (prev == null) node.searchSucceeded (this);
+ if (prev == null) node.increaseSearchRate();
else prev.sendMessage (new InsertReply (id));
}
@@ -148,7 +150,10 @@
searchState = COMPLETED;
node.cacheChk (key);
node.storeChk (key);
- if (prev == null) node.log (this + " completed");
+ if (prev == null) {
+ if (transferred) node.log (this + " succeeded");
+ else node.log (this + " failed (no xfers)");
+ }
else prev.sendMessage (new TransfersCompleted (id));
node.removeMessageHandler (id);
}
@@ -164,13 +169,25 @@
return new String ("CHK insert (" + id + "," + key + ")");
}
+ // Overrides MessageHandler.searchTimeout() to suppress failure message
+ protected void searchTimeout (Peer p)
+ {
+ if (p != next) return; // We've already moved on to another peer
+ if (searchState != ACCEPTED) return;
+ node.log (this + " search timeout for " + p);
+ p.localRejectedOverload(); // Back off from p
+ // Tell the sender to slow down
+ if (prev == null) node.decreaseSearchRate(); // Don't fail yet
+ else prev.sendMessage (new RejectedOverload (id, false));
+ finish();
+ }
+
// Event callback
private void dataTimeout()
{
if (inState != STARTED) return;
node.log (this + " data timeout from " + prev);
- if (prev == null) node.log (this + " failed (xfer start)");
- else prev.sendMessage (new TransfersCompleted (id));
+ if (prev != null) prev.sendMessage (new TransfersCompleted(id));
reallyFinish();
}
@@ -179,8 +196,7 @@
{
if (inState != TRANSFERRING) return;
node.log (this + " transfer timeout from " + prev);
- if (prev == null) node.log (this + " failed (xfer end)");
- else prev.sendMessage (new TransfersCompleted (id));
+ if (prev != null) prev.sendMessage (new TransfersCompleted(id));
reallyFinish();
}
Modified:
trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkRequestHandler.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkRequestHandler.java
2006-11-18 09:11:16 UTC (rev 10982)
+++ trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkRequestHandler.java
2006-11-18 12:53:43 UTC (rev 10983)
@@ -24,7 +24,7 @@
else if (m instanceof RejectedLoop)
handleRejectedLoop ((RejectedLoop) m);
else if (m instanceof RejectedOverload)
- handleOverload ((RejectedOverload) m, src);
+ handleRejectedOverload ((RejectedOverload) m);
else if (m instanceof RouteNotFound)
handleRouteNotFound ((RouteNotFound) m);
else if (m instanceof DataNotFound)
@@ -44,7 +44,10 @@
// If we have all the blocks and the headers, cache the data
if (blocksReceived == 32) {
node.cacheChk (key);
- if (prev == null) node.searchSucceeded (this);
+ if (prev == null) {
+ node.log (this + "succeeded");
+ node.increaseSearchRate();
+ }
finish();
}
// Wait for the transfer to complete (FIXME: check real timeout)
@@ -65,7 +68,10 @@
// If we have all the blocks and the headers, cache the data
if (blocksReceived == 32 && searchState == TRANSFERRING) {
node.cacheChk (key);
- if (prev == null) node.searchSucceeded (this);
+ if (prev == null) {
+ node.log (this + " succeeded");
+ node.increaseSearchRate();
+ }
finish();
}
}
Modified: trunk/apps/load-balancing-sims/phase7/sim/handlers/MessageHandler.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/sim/handlers/MessageHandler.java
2006-11-18 09:11:16 UTC (rev 10982)
+++ trunk/apps/load-balancing-sims/phase7/sim/handlers/MessageHandler.java
2006-11-18 12:53:43 UTC (rev 10983)
@@ -120,14 +120,15 @@
forwardSearch();
}
- protected void handleOverload (RejectedOverload ro, Peer src)
+ protected void handleRejectedOverload (RejectedOverload ro)
{
if (ro.local) {
ro.local = false;
- src.localRejectedOverload(); // Back off
- if (src == next) forwardSearch(); // Try another peer
+ next.localRejectedOverload(); // Back off
+ forwardSearch(); // Try another peer
}
- if (prev == null) node.reduceSearchRate (this);
+ // Tell the sender to slow down
+ if (prev == null) node.decreaseSearchRate();
else prev.sendMessage (ro); // Forward the message
}
@@ -147,7 +148,7 @@
node.log (this + " accepted timeout for " + p);
p.localRejectedOverload(); // Back off from p
// Tell the sender to slow down
- if (prev == null) node.reduceSearchRate (this);
+ if (prev == null) node.decreaseSearchRate();
else prev.sendMessage (new RejectedOverload (id, false));
// Try another peer
forwardSearch();
@@ -161,9 +162,11 @@
node.log (this + " search timeout for " + p);
p.localRejectedOverload(); // Back off from p
// Tell the sender to slow down
- if (prev == null) node.reduceSearchRate (this);
+ if (prev == null) {
+ node.log (this + " failed (search)");
+ node.decreaseSearchRate();
+ }
else prev.sendMessage (new RejectedOverload (id, false));
- if (prev == null) node.log (this + " failed (search)");
finish();
}
Modified:
trunk/apps/load-balancing-sims/phase7/sim/handlers/SskInsertHandler.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/sim/handlers/SskInsertHandler.java
2006-11-18 09:11:16 UTC (rev 10982)
+++ trunk/apps/load-balancing-sims/phase7/sim/handlers/SskInsertHandler.java
2006-11-18 12:53:43 UTC (rev 10983)
@@ -57,7 +57,7 @@
else if (m instanceof RejectedLoop)
handleRejectedLoop ((RejectedLoop) m);
else if (m instanceof RejectedOverload)
- handleOverload ((RejectedOverload) m, src);
+ handleRejectedOverload ((RejectedOverload) m);
else if (m instanceof RouteNotFound)
handleRouteNotFound ((RouteNotFound) m);
else if (m instanceof SskDataFound)
@@ -99,14 +99,20 @@
private void handleInsertReply (InsertReply ir)
{
if (searchState != ACCEPTED) node.log (ir + " out of order");
- if (prev == null) node.searchSucceeded (this);
+ if (prev == null) {
+ node.log (this + " succeeded");
+ node.increaseSearchRate();
+ }
else prev.sendMessage (ir); // Forward the message
finish();
}
protected void sendReply()
{
- if (prev == null) node.searchSucceeded (this);
+ if (prev == null) {
+ node.log (this + " succeeded");
+ node.increaseSearchRate();
+ }
else prev.sendMessage (new InsertReply (id));
}
Modified:
trunk/apps/load-balancing-sims/phase7/sim/handlers/SskRequestHandler.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/sim/handlers/SskRequestHandler.java
2006-11-18 09:11:16 UTC (rev 10982)
+++ trunk/apps/load-balancing-sims/phase7/sim/handlers/SskRequestHandler.java
2006-11-18 12:53:43 UTC (rev 10983)
@@ -27,7 +27,7 @@
else if (m instanceof RejectedLoop)
handleRejectedLoop ((RejectedLoop) m);
else if (m instanceof RejectedOverload)
- handleOverload ((RejectedOverload) m, src);
+ handleRejectedOverload ((RejectedOverload) m);
else if (m instanceof RouteNotFound)
handleRouteNotFound ((RouteNotFound) m);
else if (m instanceof DataNotFound)
@@ -44,7 +44,10 @@
if (searchState != ACCEPTED) node.log (df + " out of order");
dataFound = df;
if (pubKey == null) return; // Keep waiting
- if (prev == null) node.searchSucceeded (this);
+ if (prev == null) {
+ node.log (this + " succeeded");
+ node.increaseSearchRate();
+ }
else {
prev.sendMessage (dataFound);
if (needPubKey) prev.sendMessage (pubKey);
@@ -59,7 +62,10 @@
if (searchState != ACCEPTED) node.log (pk + " out of order");
pubKey = pk;
if (dataFound == null) return; // Keep waiting
- if (prev == null) node.searchSucceeded (this);
+ if (prev == null) {
+ node.log (this + " succeeded");
+ node.increaseSearchRate();
+ }
else {
prev.sendMessage (dataFound);
if (needPubKey) prev.sendMessage (pubKey);
