Update of /cvsroot/freenet/freenet/src/freenet/node/states/request
In directory sc8-pr-cvs1:/tmp/cvs-serv32518/src/freenet/node/states/request
Modified Files:
AwaitingInsert.java AwaitingStoreData.java DataPending.java
InsertPending.java Pending.java ReceivingInsert.java
ReceivingReply.java RequestState.java SendingReply.java
TransferInsert.java TransferInsertPending.java
TransferReply.java
Log Message:
6163: Merge NGRouting
* Routing table switch: new config option routingTableImpl - "classic" or "ng"
* New diagnostics variables: normalizedSuccessTime, successSearchTime,
successTransferRate, requestSuccessRatio, requestFailureRoutingOrNotRatio,
routingSuccessRatio. These are reported by both RT modes, so should be usable for
comparative purposes.
* Significant changes to Node Status Servlet, including a page with several graphs for
each node in the RT.
* Significant changes to ConnectionOpener. Fixes to races that were causing
RouteNotFound errors, a toString(), much more and better logging.
* Make a distinction between failed caching a DataReply due to IOE in the store and
due to IOE on the stream when caching a DataReply. Use this in the code to make sure
we respond correctly w.r.t. routing (it's more important in NGRouting).
* RoutingTable.route() now takes extra params: HTL, size, whether it is an insert.
Added reportConnectionSuccess, reportConnectionFailure, used by ConnectionOpener to
tell RT about connections.
* Major changes to the existing ResponseTimeEstimator class. Changed logic, changed
serialization, changed interface, just about everything. Now implements the generic
TimeEstimator, knows about time versus transfer rate versus probability, wraps around
properly, has some significant algorithm changes.
* Change Routing interface significantly. Several new me methods, old ones have been
given new arguments. Caller must for stats eventually call terminate(), or another
terminal method (transferSuccess, dataNotFound). New base class, TerminatableRouting
(implements Routing, base of NGRouting and TreeRouting), handles most of the stats.
* Add code all over states/ (mostly request, but also the others) for notifying the
Routing of timings and status, terminating the Routing, some reporting bugfixes, get
rid of the old "NGROUTING" debug log diagnostics.
* Lots of completely new code for NGRouting, mostly in freenet/node/rt/
Misc fixes etc
* Fix a deadlock in ConnectionHandler
* Iakin fix for 1.4.2 and later changed static initialization behaviour w.r.t.
Core.params.
* Added messageSuccessRatio, tracks proportion of messages successfully sent
* Fix some NullPointerExceptions
* Spelling fixes to descriptions of diagnostic variables by Edward J. Huff
* Fixes to AbstractSelectorLoop.queueClose*, connection close notifications were being
missed.
* Catch all throwables thrown by process() processing maintenance queue
Minor/future
* Notes on inserts (attacks, NGRouting)
* Give the ref to the GPL on the RNF page a filename
* Even more logging improvements
* More minor reindenting/restyling
Index: AwaitingInsert.java
===================================================================
RCS file:
/cvsroot/freenet/freenet/src/freenet/node/states/request/AwaitingInsert.java,v
retrieving revision 1.6
retrieving revision 1.7
diff -u -r1.6 -r1.7
--- AwaitingInsert.java 18 Nov 2002 22:07:43 -0000 1.6
+++ AwaitingInsert.java 30 Aug 2003 23:16:53 -0000 1.7
@@ -34,7 +34,7 @@
if(ni != null) ni.cancel();
fail(n, "State lost while waiting for your DataInsert");
}
-
+
public State receivedMessage(Node n, DataInsert dim) throws StateException {
if (!fromOrigPeer(dim)) {
throw new BadStateException("DataInsert from the wrong peer!");
@@ -42,25 +42,23 @@
if(ni != null) ni.cancel();
try {
receivingData = dim.cacheData(n, searchKey);
- }
- catch (KeyCollisionException e) {
+ } catch (KeyCollisionException e) {
// we've already sent the Accepted so we should try..
dim.eatData(n);
scheduleRestart(n, 0);
// this bloody well ought to work
return new DataPending(this);
- }
- catch (IOException e) {
+ } catch (IOException e) {
fail(n, "I/O error receiving insert");
n.logger.log(this, "Failed to cache insert on chain "
- + Long.toHexString(id), e, Logger.ERROR);
+ + Long.toHexString(id), e, Logger.ERROR);
return new RequestDone(this);
}
receivingData.schedule(n);
return new ReceivingInsert(this);
}
-
+
public State receivedMessage(Node n, NoInsert ni) throws StateException {
if (this.ni != ni) {
throw new BadStateException("Not my NoInsert: "+ni);
@@ -68,10 +66,9 @@
n.logger.log(this, "Did not receive expected DataInsert on chain "
+ Long.toHexString(id), Logger.MINOR);
fail(n, "DataInsert never received");
-
return new RequestDone(this);
- }
-
+ }
+
public State receivedMessage(Node n, QueryAborted q) throws StateException {
Core.logger.log(this, "Aborted AwaitingInsert", Core.logger.DEBUG);
if(ni != null) ni.cancel();
Index: AwaitingStoreData.java
===================================================================
RCS file:
/cvsroot/freenet/freenet/src/freenet/node/states/request/AwaitingStoreData.java,v
retrieving revision 1.24
retrieving revision 1.25
diff -u -r1.24 -r1.25
--- AwaitingStoreData.java 12 Aug 2003 02:41:57 -0000 1.24
+++ AwaitingStoreData.java 30 Aug 2003 23:16:53 -0000 1.25
@@ -56,10 +56,10 @@
// this is pretty much the same as if it timed out for a DataRequest,
// but for an insert we need to go back to TransferInsertPending
if (xferIns == null) {
+ terminateRouting(false, true);
relayStoreData(n, null);
return new RequestDone(this);
- }
- else {
+ } else {
if (nosd != null) nosd.cancel();
return xferIns.receivedMessage(n, qr);
}
@@ -69,6 +69,7 @@
if (this.nosd != nosd) {
throw new BadStateException("Not my NoStoreData: "+nosd);
}
+ terminateRouting(false, true);
relayStoreData(n, null);
return new RequestDone(this);
}
@@ -82,6 +83,7 @@
// Update global network load estimate stats.
n.loadStats.storeTraffic(sd.dataSource(), sd.requestsPerHour());
+ terminateRouting(true, true);
relayStoreData(n, sd);
return new RequestDone(this);
@@ -90,6 +92,7 @@
public final void lost(Node n) {
Core.diagnostics.occurrenceCounting("lostAwaitingStoreData", 1);
// just like timing out with the NoStoreData
+ terminateRouting(false, false);
relayStoreData(n, null);
}
Index: DataPending.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/DataPending.java,v
retrieving revision 1.15
retrieving revision 1.16
diff -u -r1.15 -r1.16
--- DataPending.java 14 Aug 2003 02:06:48 -0000 1.15
+++ DataPending.java 30 Aug 2003 23:16:53 -0000 1.16
@@ -46,17 +46,17 @@
try {
checkFailureTable(n);
super.receivedQueryRejected(n, qr);
- }
- catch (EndOfRouteException e) {
+ } catch (EndOfRouteException e) {
dataNotFound(n);
+ terminateRouting(false, true); // because routing produced the QR
return new RequestDone(this);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
+ // Might not be a RequestDone - terminate WHEN THROWING
return rae.state;
}
return new DataPending(this);
}
-
+
public State receivedMessage(Node n, RequestInitiator ri) throws StateException {
if (this.ri == null || ri != this.ri) {
throw new BadStateException("Not my request initiator: "+ri+" for "+this);
@@ -67,19 +67,18 @@
try {
checkFailureTable(n);
super.receivedRequestInitiator(n, ri);
- }
- catch (EndOfRouteException e) {
+ } catch (EndOfRouteException e) {
dataNotFound(n);
+ terminateRouting(false, true);
return new RequestDone(this);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
return rae.state;
}
return new DataPending(this);
}
// the rest don't...
-
+
public State receivedMessage(Node n, DataRequest dr) {
try {
checkFailureTable(n);
@@ -89,7 +88,7 @@
return rae.state;
}
}
-
+
public State receivedMessage(Node n, QueryRestarted qr) throws StateException {
try {
super.receivedQueryRestarted(n, qr);
@@ -99,12 +98,12 @@
}
return this;
}
-
+
public State receivedMessage(Node n, Accepted a) throws StateException {
super.receivedAccepted(n, a);
return this;
}
-
+
public State receivedMessage(Node n, DataReply dr) throws StateException {
return super.receivedDataReply(n, dr);
}
@@ -124,24 +123,27 @@
long toq = Math.min(dnf.timeOfQuery(), // not stupid...
System.currentTimeMillis());
+ routes.dataNotFound(hopsToLive); // is terminal
dataNotFound(n, toq, true);
// Add this key to the FailureTable
if (!n.ds.contains(searchKey)) // that sort of sucks...
n.ft.failedToFind(searchKey, hopsToLive, toq);
// well, technically they did their job...
routes.routeSucceeded(dnf.source.isCached());
-
return new RequestDone(this);
}
private final void checkFailureTable(Node n) throws RequestAbortException {
long toq = n.ft.shouldFail(searchKey, hopsToLive);
if ((origPeer != null) && toq > 0) {
- dataNotFound(n, toq, false);
+ dataNotFound(n, toq, true);
+ terminateRouting(false, false);
+ // It certainly wasn't *THIS* Routing's fault
+
throw new RequestAbortException(new RequestDone(this));
}
}
-
+
private final void dataNotFound(Node n) {
dataNotFound(n, System.currentTimeMillis(), false);
}
@@ -156,6 +158,10 @@
n.logger.log(this, "Failed to reply with DataNotFound: "+e+
" for "+this, Logger.MINOR);
}
+ }
+
+ protected boolean isInsert() {
+ return false;
}
}
Index: InsertPending.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/InsertPending.java,v
retrieving revision 1.17
retrieving revision 1.18
diff -u -r1.17 -r1.18
--- InsertPending.java 19 Jun 2003 13:31:20 -0000 1.17
+++ InsertPending.java 30 Aug 2003 23:16:53 -0000 1.18
@@ -20,11 +20,19 @@
DataInsert dim; // cache it for sending upstream
long dimRecvTime, // when we got the DataInsert
- checkTime; // when we got the Accepted/InsertReply
-
+ checkTime, // when we got the Accepted/InsertReply
+ startedSendTime, // when we started the SendData
+ insertReplyTime = -1; // when we got the InsertReply
+
boolean approved = false; // got an InsertReply?
+ public String toString() {
+ String s = super.toString();
+ return s + ", "+(approved ? "approved" : "not-approved")+
+ ", insertReplyTime="+insertReplyTime;
+ }
+
/**
* For new requests.
*/
@@ -39,6 +47,7 @@
InsertPending(InsertPending ancestor) {
super(ancestor);
ni = ancestor.ni;
+ insertReplyTime = ancestor.insertReplyTime;
}
@@ -58,11 +67,11 @@
public State receivedMessage(Node n, QueryRejected qr) throws StateException {
try {
super.receivedQueryRejected(n, qr);
- }
- catch (EndOfRouteException e) {
+ } catch (EndOfRouteException e) {
+ // endRoute will either terminate or go to AwaitingInsert, either way the
route is ended, and it's the route's fault
+ terminateRouting(false, true);
return endRoute(n);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
cancelNoInsert();
// we are either going to SendingReply or RequestDone with no route found
return rae.state;
@@ -81,13 +90,14 @@
}
try {
super.receivedRequestInitiator(n, ri);
- }
- catch (EndOfRouteException e) {
+ } catch (EndOfRouteException e) {
+ // see above
+ terminateRouting(false,true);
return endRoute(n);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
cancelNoInsert();
// this is going to RequestDone with no route found
+ // Should be terminate()d at throw time, not at catch time
return rae.state;
}
return new InsertPending(this);
@@ -118,6 +128,8 @@
}
fail(n, "DataInsert never received");
queryAborted(n);
+ // Not our fault, this is from the requester, so don't tell Routing
+ terminateRouting(false, false);
return new RequestDone(this);
}
@@ -128,7 +140,7 @@
}
cancelNoInsert();
queryAborted(n, qf);
-
+ terminateRouting(false, false);
return new RequestDone(this);
}
@@ -144,34 +156,33 @@
try {
receivingData = dim.cacheData(n, searchKey);
- }
- catch (KeyCollisionException e) {
+ } catch (KeyCollisionException e) {
// we've already sent the Accepted so we should try..
dim.eatData(n);
// propagate QueryAborted upstream in case they
// are waiting on DataInsert
queryAborted(n);
-
+ terminateRouting(false, false);
+ // Won't route again!
try {
searchData(n);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
return rae.state;
}
-
+
// well damn, it's gone already
fail(n, "I/O error receiving insert");
n.logger.log(this, "Failed to find data after key collision for "+this,
Logger.NORMAL);
- return new RequestDone(this);
- }
- catch (IOException e) {
+ return new RequestDone(this);
+ } catch (IOException e) {
dim.drop(n);
fail(n, "I/O error receiving insert");
n.logger.log(this, "Failed to cache insert for " + this,
e, Logger.ERROR);
queryAborted(n);
+ terminateRouting(false, false);
return new RequestDone(this);
}
@@ -185,20 +196,17 @@
relayInsert(n);
return new TransferInsert(this);
- }
- else {
+ } else {
KeyInputStream doc = receivingData.getKeyInputStream();
doc.setParent(id, n.ticker().getMessageHandler(),
"Set to TransferInsertPending for "+
searchKey);
return new TransferInsertPending(this, doc);
}
- }
- catch (RequestAbortException e) {
+ } catch (RequestAbortException e) {
// immediate restart, or failure
return e.state;
- }
- catch (IOException e) {
+ } catch (IOException e) {
receivingData.cancel();
fail(n, "I/O error receiving insert");
if (e instanceof BufferException)
@@ -208,30 +216,32 @@
n.logger.log(this, "Failed to cache insert for "+this,
e, Logger.ERROR);
queryAborted(n);
+ terminateRouting(false, false);
return new RequestDone(this);
- }
- finally {
+ } finally {
receivingData.schedule(n);
}
}
-
+
public State receivedMessage(Node n, DataReply dr) throws StateException {
State ret = super.receivedDataReply(n, dr);
+ // super will deal with Routing
if (this != ret) cancelNoInsert();
return ret;
}
-
+
public State receivedMessage(Node n, Accepted a) throws StateException {
if (!accepted)
checkTime = System.currentTimeMillis();
super.receivedAccepted(n, a);
return this;
}
-
+
public State receivedMessage(Node n, InsertReply ir) throws StateException {
if (!fromLastPeer(ir)) {
throw new BadStateException("InsertReply from wrong peer!");
}
+ insertReplyTime = System.currentTimeMillis();
if (!approved) {
if (routedTime > 0)
Core.diagnostics.occurrenceContinuous("hopTime",
@@ -244,20 +254,21 @@
approved = true;
routes.routeAccepted();
try {
+ // FIXME: subtract time to send the insertReply from the
insertReplyTime
+ // Or make insertReply asynchronous (with feedback...)
insertReply(n);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
cancelNoInsert();
- // send failed to iriginator..
+ // send failed to originator..
return rae.state;
}
}
return this;
}
-
-
+
+
//=== support methods ======================================================
-
+
final void cancelNoInsert() {
if (ni != null) {
ni.cancel();
@@ -267,17 +278,19 @@
private final State endRoute(Node n) {
cancelRestart();
- logFailure(n);
+// logFailure(n);
+ terminateRouting(false, false); // just in case
+ // Nobody to route it to, so terminate any routing
+
try {
insertReply(n);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
cancelNoInsert();
return rae.state;
}
return new AwaitingInsert(this);
}
-
+
protected State publicEndRoute(Node n) {
return endRoute(n);
} // FIXME!
@@ -297,26 +310,26 @@
}
}
}
-
+
final void insertReply(Node n) throws RequestAbortException {
try {
ft.insertReply(n,
Core.storeDataTime(hopsToLive,
searchKey.getExpectedTransmissionLength()));
- }
- catch (CommunicationException e) {
+ } catch (CommunicationException e) {
if (receivingData != null) receivingData.cancel();
queryAborted(n);
n.logger.log(this,
"Failed to send back InsertReply, dropping for "+
this, e, Logger.MINOR);
+ terminateRouting(false, false);
throw new RequestAbortException(new RequestDone(this));
}
}
/** Schedules the SendData state to relay the data upstream
- * @throws RequestAbortException if there is a failure
- */
+ * @throws RequestAbortException if there is a failure
+ */
void relayInsert(Node n) throws RequestAbortException {
try {
KeyInputStream doc = receivingData.getKeyInputStream();
@@ -324,30 +337,29 @@
"InsertPending.relayInsert");
try {
relayInsert(n, doc);
- }
- catch (RequestAbortException e) {
+ } catch (RequestAbortException e) {
doc.close();
throw e;
}
- }
- catch (IOException e) {
+ } catch (IOException e) {
receivingData.cancel();
fail(n, "I/O error receiving insert");
queryAborted(n);
- if (e instanceof BufferException) {
- n.logger.log(this, "Failed to read data from store: "+e+
- " for "+this, Logger.MINOR);
- }
- else {
+// if (e instanceof BufferException) {
+// n.logger.log(this, "Failed to read data from store: "+e+
+// " for "+this, Logger.MINOR);
+// } else {
n.logger.log(this, "Failed to read data from store for "+
this, e, Logger.ERROR);
- }
+// }
+ terminateRouting(false, false);
throw new RequestAbortException(new RequestDone(this));
}
}
-
+
void relayInsert(Node n, KeyInputStream doc) throws RequestAbortException {
OutputStream out;
+ startedSendTime = System.currentTimeMillis();
try {
receivedTime = -2; // receivedTime on the next message will be meaningless
if(logDEBUG)
@@ -358,8 +370,7 @@
out = ch.sendMessage(dim);
if(logDEBUG)
n.logger.log(this, "Sent message for "+this+" on "+ch, Logger.DEBUG);
- }
- catch (CommunicationException e) {
+ } catch (CommunicationException e) {
// restart right away
scheduleRestart(n, 0);
throw new RequestAbortException(this);
@@ -374,6 +385,10 @@
sendingData.schedule(n);
if(logDEBUG) n.logger.log(this, "Scheduled "+sendingData+" for "+this,
Logger.DEBUG);
+ }
+
+ protected boolean isInsert() {
+ return true;
}
}
Index: Pending.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/Pending.java,v
retrieving revision 1.48
retrieving revision 1.49
diff -u -r1.48 -r1.49
--- Pending.java 26 Jul 2003 21:36:29 -0000 1.48
+++ Pending.java 30 Aug 2003 23:16:53 -0000 1.49
@@ -58,7 +58,7 @@
long searchDataRoutingTime = -1;
public String toString() {
- return super.toString()+", routedTime="+routedTime;
+ return super.toString()+", routedTime="+routedTime+", replyTime="+replyTime;
}
/** Pending states may be created from scratch.
@@ -90,6 +90,7 @@
*/
public void lost(Node n) {
Core.diagnostics.occurrenceCounting("lostRequestState", 1);
+ terminateRouting(false, false);
fail(n, "Request state lost due to overflow for "+this);
}
@@ -97,10 +98,10 @@
* (note that unrecognized fields are lost)
*/
abstract Request createRequest(FieldSet otherFields, NodeReference ref);
-
+
//=== message handling =====================================================
-
+
void receivedStoreData(Node n, StoreData sd) throws BadStateException {
if (!fromLastPeer(sd)) {
throw new BadStateException("StoreData from the wrong peer!");
@@ -112,18 +113,18 @@
void receivedQueryRestarted(Node n, QueryRestarted qr) throws BadStateException,
RequestAbortException {
- acceptedTime = -1; // this time is no longer meaningful
- receivedTime = -2;
-
if (!fromLastPeer(qr)) {
throw new BadStateException("QueryRestarted from the wrong peer!");
}
+ acceptedTime = -1; // this time is no longer meaningful
+ receivedTime = -2;
+
cancelRestart();
long timeout = Core.hopTime(hopsToLive);
relayRestarted(n, timeout);
scheduleRestart(n, timeout);
}
-
+
/**
* Note the EndOfRouteException and RequestAbortException.
* Must be called by subclass as part of the implementation
@@ -135,12 +136,12 @@
if (!fromLastPeer(qr)) {
throw new BadStateException("QueryRejected from the wrong peer!");
}
-
+
gotTime = System.currentTimeMillis();
if(receivedTime >= -1) {
receivedTime = qr.getReceivedTime(); // measure routing time again
- if(logDEBUG) n.logger.log(this, "Remeasured receivedTime: "+receivedTime,
- Logger.DEBUG);
+ if(logDEBUG) n.logger.log(this, "Remeasured receivedTime: "+receivedTime+
+ " for "+this, Logger.DEBUG);
}
if (receivedTime < 1000*1000*1000*1000) {
if(logDEBUG && receivedTime > -1)
@@ -155,18 +156,24 @@
new Exception("debug"), Logger.DEBUG);
}
- logFailure(n, gotTime);
+// logFailure(n, gotTime);
++rejected;
-
+
// reduce our HTL like the man sez, but by at least 1
hopsToLive = (int) Math.min(hopsToLive-1, qr.hopsToLive);
if(logDEBUG)
- n.logger.log(this, "Rejected count: "+rejected+", current hopsToLive: "+
+ n.logger.log(this, "Rejected count: "+rejected+
+ ", current hopsToLive: "+
hopsToLive+" for "+this, Logger.DEBUG);
- routes.queryRejected(qr.source.isCached(), qr.attenuation);
+ if(routes == null) {
+ n.logger.log(this, "Already terminated in "+this+
+ ".receivedQueryRejected!", Logger.NORMAL);
+ } else {
+ routes.queryRejected(qr.source.isCached(), qr.attenuation);
+ }
long toldRTTime = System.currentTimeMillis();
if(logDEBUG)
n.logger.log(this, "Time to inform RT: "+toldRTTime+" for "+
@@ -197,15 +204,22 @@
*/
void receivedRequestInitiator(Node n, RequestInitiator ri) throws
BadStateException,
RequestAbortException,
-
EndOfRouteException {
+
EndOfRouteException {
// it must be my own RequestInitiator
if (this.ri == null || this.ri != ri) {
throw new BadStateException("Not my request initiator: "+ri);
}
-
+
gotTime = System.currentTimeMillis();
- logFailure(n, gotTime);
+// logFailure(n, gotTime);
+ if(routedTime > 0) {
+ if(routes == null)
+ Core.logger.log(this, "routes NULL in
"+this+".receivedRequestInitiator",
+ Logger.ERROR);
+ else
+ routes.searchFailed(gotTime - routedTime);
+ }
if(receivedTime >= -1) {
receivedTime = ri.startTime();
// message receive time or the
@@ -241,30 +255,13 @@
n.logger.log(this, "searchDataRoutingTime took "+
(searchDataTime - regotTime)+"ms on "+
this+"!", Core.logger.MINOR);
- if (origPeer != null)
- {
- if(logDEBUG)
- Core.logger.log(this, "About to route() at "+
- System.currentTimeMillis()+" for "+
- this, Logger.DEBUG);
- routes = n.rt.route(searchKey, false);
- }
- else
- {
- // Route to random first hop, copied from NewAnnouncement
- long l = n.randSource.nextLong();
- Key k = new Key(new byte[] {
- (byte) l, (byte) (l >> 8), (byte) (l >> 16),
- (byte) (l >> 24), (byte) (l >> 32), (byte) (l >> 40),
- (byte) (l >> 48), (byte) (l >> 56)
- });
- if(logDEBUG)
- Core.logger.log(this, "About to route() at "+
- System.currentTimeMillis()+
- " for "+this+", random routed as "+k,
- Logger.DEBUG);
- routes = n.rt.route(k, true);
- }
+ if(logDEBUG)
+ Core.logger.log(this, "About to route() at "+
+ System.currentTimeMillis()+" for "+
+ this, Logger.DEBUG);
+ routes = n.rt.route(searchKey, hopsToLive,
+ searchKey.getExpectedTransmissionLength(),
+ false, isInsert());
gotRouteTime = System.currentTimeMillis();
n.diagnostics.occurrenceContinuous("getRouteTime", gotRouteTime -
searchDataTime);
if((gotRouteTime - searchDataTime) > 1000)
@@ -296,13 +293,13 @@
// decrease contact probability for node that didn't reply
if (lastPeer != null)
routes.timedOut();
-
+
// Don't allow indefinite restarts.
//
// Note:
// sendOn will throw an EndOfRouteException below
// if hopsToLive hits 0.
-
+
if (--hopsToLive > 0) {
// send QueryRestarted to initiating chain
relayRestarted(n, Core.hopTime(hopsToLive + 1));
@@ -321,7 +318,9 @@
// Requests should actually not carry unknown fields...
sendOn(n, createRequest(null, n.myRef), true);
}
-
+
+ protected abstract boolean isInsert();
+
/**
* Because method selection on superclasses doesn't work,
* implementation must call this from a method with the actual
@@ -332,10 +331,11 @@
if(logDEBUG)
n.logger.log(this, "Backtracking", Logger.DEBUG);
try {
- n.sendMessage(
- new QueryRejected(id, hopsToLive, "Looped request", r.otherFields),
- r.getSource(), 0
- );
+ Message m = new QueryRejected(id, hopsToLive, "Looped request",
r.otherFields);
+ ConnectionHandler ch = n.makeConnection(r.getSource());
+ RequestSendCallback cb = new RequestSendCallback("QueryRejected (looped
request)"+
+ " for "+this, n, this);
+ ch.sendMessageAsync(m,cb);
} catch (CommunicationException e) {
n.logger.log(this, "Failed to send backtrack reply: "+e+" for "+this,
Logger.MINOR);
@@ -373,43 +373,30 @@
accepted = true; // more or less..
- routes.routeAccepted();
+ if(acceptedTime <= 0) routes.routeAccepted();
cancelRestart();
try {
receivingData = dr.cacheData(n, searchKey);
replyTime = System.currentTimeMillis();
// replyTime must be set AFTER verifying Storables
- if(routedTime > 0 && logDEBUG) {
- n.logger.log(this, "NGROUTING: Key "+searchKey+" took "+
- (replyTime - routedTime)+" for "+hopsToLive,
- Logger.DEBUG);
-
- } else {
- if(logDEBUG)
- n.logger.log(this, "NGROUTING: Key "+searchKey+
- ": no valid routedTime", Logger.DEBUG);
- }
n.ft.remove(searchKey); // remove if in FT.
- } catch (DataNotValidIOException e) {
- dr.drop(n);
- n.logger.log(this,
- "Got DNV: "+Presentation.getCBdescription(e.getCode())+
- " for "+this,
- Logger.MINOR);
- scheduleRestart(n, 0); // schedule restart now
- return this; // back to pending
- }
- catch (KeyCollisionException e) {
+ } catch (StoreIOException e) {
+ // Our fault
+ terminateRouting(false, false);
+ dr.drop(n);
+ fail(n, "I/O error storing DataReply");
+ return new RequestDone(this);
+ } catch (KeyCollisionException e) {
// oh well, try to go to SendingReply
if(logDEBUG)
n.logger.log(this, "Got KeyCollisionException on "+
this, Logger.DEBUG);
+ replyTime = -4;
dr.drop(n);
try {
searchData(n);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
if(logDEBUG)
n.logger.log(this, "RAE in got KCE on "+this, rae,
Logger.DEBUG);
@@ -421,15 +408,23 @@
"Failed to find data after key collision on "+
this+" while caching DataReply",
Logger.NORMAL);
+ terminateRouting(false, false); // cache failure? anyway, node not routing
return new RequestDone(this);
- }
- catch (IOException e) {
+ } catch (IOException e) {
+ // Their fault
dr.drop(n);
- fail(n, "I/O error replying with data");
- n.logger.log(this, "I/O error caching DataReply "+this,
- e, Logger.ERROR);
- return new RequestDone(this);
- }
+ routes.verityFailed();
+ // The transfer has not started yet
+ if(e instanceof DataNotValidIOException) {
+ DataNotValidIOException de = (DataNotValidIOException)e;
+ n.logger.log(this, "Got DNV: "+
+ Presentation.getCBdescription(de.getCode())+
+ " for "+this, Logger.MINOR);
+ }
+ scheduleRestart(n, 0); // schedule restart now
+ replyTime = -3;
+ return this; // back to pending
+ }
try {
KeyInputStream kin = receivingData.getKeyInputStream();
@@ -446,8 +441,7 @@
if (!worked)
kin.close();
}
- }
- catch (IOException e) {
+ } catch (IOException e) {
// couldn't get the KeyInputStream
fail(n, "I/O error replying with data");
if (e instanceof BufferException) {
@@ -458,30 +452,35 @@
n.logger.log(this, "I/O error getting KeyInputStream for "+
this, e, Logger.ERROR);
}
+ // Our fault, or upstream's fault
+ // Routing is still running and valid
return new ReceivingReply(this);
- }
- catch (CommunicationException e) {
+ } catch (CommunicationException e) {
+ // Our fault, or upstream's fault
+ // So don't tell the Routing
n.logger.log(this, "Error replying to peer: "+e+" for "+this,
Logger.MINOR);
return new ReceivingReply(this);
- }
- finally {
+ } finally {
receivingData.schedule(n);
}
-
+
return new TransferReply(this);
}
-
+
//=== support methods ======================================================
-
+
private void sendOn(Node n, Request r, boolean isFirst)
throws EndOfRouteException, RequestAbortException {
loggedResult = false;
- if (hopsToLive <= 0)
+ if (hopsToLive <= 0) {
+ terminateRouting(false, true /* probably! */);
throw new EndOfRouteException("Reached last node in route");
+ }
+
Peer addr = null;
NodeReference nr;
@@ -491,7 +490,11 @@
if(logDEBUG)
n.logger.log(this, "Still trying to route "+searchKey+"("+
this+"), attempt "+attemptCount, Logger.DEBUG);
- nr = routes.getNextRoute();
+ if(routes == null) {
+ nr = null; // terminated
+ n.logger.log(this, "Already terminated but trying to route!: "+this,
+ Logger.NORMAL);
+ } else nr = routes.getNextRoute();
if(logDEBUG) n.logger.log(this, "Got next route for "+this,
Logger.DEBUG);
long stillInSendOnTime = System.currentTimeMillis();
@@ -511,6 +514,7 @@
if(logDEBUG)
n.logger.log(this, "rt exhaused for "+this,
Logger.DEBUG);
+ terminateRouting(false, true);
throw new RequestAbortException(new RequestDone(this));
} else {
if(logDEBUG)
@@ -522,7 +526,8 @@
Logger.DEBUG);
long stillStillInSendOnTime = System.currentTimeMillis();
if(isFirst && attemptCount == 0) {
- n.diagnostics.occurrenceContinuous("stillStillInSendOnTime",
stillStillInSendOnTime -
+ n.diagnostics.occurrenceContinuous("stillStillInSendOnTime",
+ stillStillInSendOnTime -
stillInSendOnTime);
}
if (addr == null) { // bad node ref
@@ -572,11 +577,12 @@
attemptCount++;
ConnectionHandler ch =
n.makeConnection(addr, n.routeConnectTimeout);
- if(logDEBUG) Core.logger.log(this, "Sending message "+r+" on "+ch+"
for "+
- this, Logger.DEBUG);
+ routes.routeConnected(ch.isCached());
+ if(logDEBUG) Core.logger.log(this, "Sending message "+r+" on "
+ +ch+" for "+this, Logger.DEBUG);
ch.sendMessage(r);
- if(logDEBUG) Core.logger.log(this, "Sent message "+r+" on "+ch+" for "+
- this, Logger.DEBUG);
+ if(logDEBUG) Core.logger.log(this, "Sent message "+r+" on "+ch
+ +" for "+this, Logger.DEBUG);
break;
} catch (CommunicationException e) {
++unreachable;
@@ -586,32 +592,27 @@
n.logger.log(this, "Routing ("+this+") failure to: "
+e.peerAddress() + " -- " + e, e,
Logger.DEBUG);
- //e, Logger.DEBUG);
if (e instanceof AuthenticationFailedException) {
- //routes.routeConnected();
routes.authFailed();
- }
- else {
+ } else {
routes.connectFailed();
}
}
}
-
+
// Count outbound Requests.
n.diagnostics.occurrenceBinomial("outboundAggregateRequests", attemptCount,
1);
if (n.outboundRequests != null) {
n.outboundRequests.incTotal(addr.getAddress().toString());
}
-
+
// Keep track of outbound requests for rate limiting.
n.outboundRequestLimit.inc();
-
+
lastPeer = addr;
- routes.routeConnected();
-
scheduleRestart(n, Core.hopTime(1)); // timeout to get the Accepted
}
-
+
private final void relayRestarted(Node n, long timeout) throws
RequestAbortException {
try {
ft.restarted(n, timeout);
@@ -620,13 +621,14 @@
n.logger.log(this,
"Couldn't restart because relaying QueryRestarted failed: "+
e+" for "+this, Logger.MINOR);
+ terminateRouting(false, false);
throw new RequestAbortException(new RequestDone(this));
}
}
-
+
/** Given an existing KeyInputStream, sets up a SendData state to
- * transfer the data back to the requester.
- */
+ * transfer the data back to the requester.
+ */
SendData sendData(Node n, KeyInputStream doc) throws CommunicationException {
if(logDEBUG) n.logger.log(this, "Sending data (,"+doc+") for "+this,
Logger.DEBUG);
@@ -667,6 +669,7 @@
doc.setParent(id, n.ticker().getMessageHandler(),
"Replying with data from store");
sendingData = sendData(n, doc);
+ terminateRouting(true, false);
long sendingDataTime = System.currentTimeMillis();
if(logDEBUG)
n.logger.log(this, "sendData() took "+
@@ -690,18 +693,19 @@
n.logger.log(this, "I/O error replying with data on "+this,
e, Logger.MINOR);
thrownTime = System.currentTimeMillis();
+ terminateRouting(false, false);
throw new RequestAbortException(new RequestDone(this));
} catch (CommunicationException e) {
n.logger.log(this, "Error replying to peer: "+e+" on "+this,
e, Logger.MINOR);
thrownTime = System.currentTimeMillis();
+ terminateRouting(false, false);
throw new RequestAbortException(new RequestDone(this));
} finally {
if (doc != null) {
try {
doc.close();
- }
- catch (IOException e) {
+ } catch (IOException e) {
n.logger.log(this,
"Failed to close KeyInputStream after failing "+
"on "+this, e, Logger.MINOR);
@@ -723,66 +727,66 @@
long endTransferTime = -1;
boolean loggedResult = false;
- public void logSuccess(Node n) {
- if(loggedResult) return;
- loggedResult = true;
- endTransferTime = System.currentTimeMillis();
-
- double transferRate = (((double)receivingData.length()) /
- ((double)(endTransferTime - replyTime))) * 1000;
- long stdFileSize;
- if(n.dir.countKeys() > 16)
- stdFileSize = (n.storeSize - n.dir.available()) / n.dir.countKeys();
- else stdFileSize = 100000;
- long expectedTime = (replyTime - routedTime) +
- (((endTransferTime - replyTime) * stdFileSize) /
- receivingData.length());
- if(logDEBUG)
- n.logger.log(this, "NGROUTING: Key "+searchKey+": search took "+
- (replyTime-routedTime)+" in "+hopsToLive+", transfer took "+
- (endTransferTime-replyTime)+" for file length "+
- receivingData.length()+": transfer rate "+transferRate+
- " bytes per second, expected total time for file of "+
- "length "+stdFileSize+": "+expectedTime+"; sent to "+
- lastPeer, Logger.DEBUG);
- }
-
- public void logFailure(Node n) {
- logFailure(n, System.currentTimeMillis());
- }
-
- public void logFailure(Node n, long endTime) {
- if(routedTime > 0) {
- if(replyTime > 0) {
- logFailedTransfer(n, endTime);
- } else {
- logFailedSearch(n, endTime);
- }
- }
- }
-
- public void logFailedTransfer(Node n) {
- endTransferTime = System.currentTimeMillis();
- logFailure(n, endTransferTime);
- }
-
- public void logFailedTransfer(Node n, long endTime) {
- if(loggedResult) return;
- loggedResult = true;
- if(logDEBUG)
- n.logger.log(this, "NGROUTING: Key "+searchKey+": search took "+
- (replyTime-routedTime)+" in "+hopsToLive+
- ", failed transfer took "+(endTransferTime - replyTime)+
- "; sent to "+lastPeer, Logger.DEBUG);
- replyTime = -1;
- }
-
- public void logFailedSearch(Node n, long endTime) {
- if(loggedResult) return;
- loggedResult = true;
- if(logDEBUG)
- n.logger.log(this, "NGROUTING: Key "+searchKey+": search failed in "+
- (endTime-routedTime)+" for "+hopsToLive+"; sent to "+
- lastPeer, new Exception("debug"), Logger.DEBUG);
- }
+// public void logSuccess(Node n) {
+// if(loggedResult) return;
+// loggedResult = true;
+// endTransferTime = System.currentTimeMillis();
+
+// double transferRate = (((double)receivingData.length()) /
+// ((double)(endTransferTime - replyTime))) * 1000;
+// long stdFileSize;
+// if(n.dir.countKeys() > 16)
+// stdFileSize = (n.storeSize - n.dir.available()) / n.dir.countKeys();
+// else stdFileSize = 100000;
+// long expectedTime = (replyTime - routedTime) +
+// (((endTransferTime - replyTime) * stdFileSize) /
+// receivingData.length());
+// if(logDEBUG)
+// n.logger.log(this, "NGROUTING: Key "+searchKey+": search took "+
+// (replyTime-routedTime)+" in "+hopsToLive+", transfer took "+
+// (endTransferTime-replyTime)+" for file length "+
+// receivingData.length()+": transfer rate "+transferRate+
+// " bytes per second, expected total time for file of "+
+// "length "+stdFileSize+": "+expectedTime+"; sent to "+
+// lastPeer, Logger.DEBUG);
+// }
+
+// public void logFailure(Node n) {
+// logFailure(n, System.currentTimeMillis());
+// }
+
+// public void logFailure(Node n, long endTime) {
+// if(routedTime > 0) {
+// if(replyTime > 0) {
+// logFailedTransfer(n, endTime);
+// } else {
+// logFailedSearch(n, endTime);
+// }
+// }
+// }
+
+// public void logFailedTransfer(Node n) {
+// endTransferTime = System.currentTimeMillis();
+// logFailure(n, endTransferTime);
+// }
+
+// public void logFailedTransfer(Node n, long endTime) {
+// if(loggedResult) return;
+// loggedResult = true;
+// if(logDEBUG)
+// n.logger.log(this, "NGROUTING: Key "+searchKey+": search took "+
+// (replyTime-routedTime)+" in "+hopsToLive+
+// ", failed transfer took "+(endTransferTime - replyTime)+
+// "; sent to "+lastPeer, Logger.DEBUG);
+// replyTime = -1;
+// }
+
+// public void logFailedSearch(Node n, long endTime) {
+// if(loggedResult) return;
+// loggedResult = true;
+// if(logDEBUG)
+// n.logger.log(this, "NGROUTING: Key "+searchKey+": search failed in "+
+// (endTime-routedTime)+" for "+hopsToLive+"; sent to "+
+// lastPeer, new Exception("debug"), Logger.DEBUG);
+// }
}
Index: ReceivingInsert.java
===================================================================
RCS file:
/cvsroot/freenet/freenet/src/freenet/node/states/request/ReceivingInsert.java,v
retrieving revision 1.18
retrieving revision 1.19
diff -u -r1.18 -r1.19
--- ReceivingInsert.java 12 Aug 2003 02:41:57 -0000 1.18
+++ ReceivingInsert.java 30 Aug 2003 23:16:53 -0000 1.19
@@ -5,6 +5,7 @@
import freenet.message.*;
import freenet.node.states.data.*;
import freenet.node.ds.KeyCollisionException;
+import freenet.node.rt.Routing;
import freenet.support.Logger;
import java.io.IOException;
@@ -67,8 +68,14 @@
// (there is still a chance of DataSource resetting)
// -1 means we don't know the rate for that ref.
n.diagnostics.occurrenceCounting("storeDataReceivingInsert", 1);
- ft.storeData(n, n.rt.route(searchKey,false).
- getNextRoute(), -1, 0, null);
+ long length = receivingData.length();
+ length = searchKey.
+ getTransmissionLength(length,
+ searchKey.getPartSize(length));
+ Routing r = n.rt.route(searchKey, hopsToLive,
+ length, false, true);
+ ft.storeData(n, r.getNextRoute(), -1, 0, null);
+ r.terminate(false, false); // duh
// We are the terminal node; we cache it
if (origPeer != null)
{
Index: ReceivingReply.java
===================================================================
RCS file:
/cvsroot/freenet/freenet/src/freenet/node/states/request/ReceivingReply.java,v
retrieving revision 1.7
retrieving revision 1.8
diff -u -r1.7 -r1.8
--- ReceivingReply.java 29 Apr 2003 00:35:39 -0000 1.7
+++ ReceivingReply.java 30 Aug 2003 23:16:53 -0000 1.8
@@ -40,12 +40,15 @@
}
public State receivedMessage(Node n, QueryRejected qr) {
+ // Transfer will be aborted
return this;
}
public State receivedMessage(Node n, DataNotFound dnf) throws BadStateException {
// technically, we might want to increase their contact probability
// if they restarted correctly and then replied with this..
+ routes.dataNotFound(hopsToLive);
+ // Transfer will be aborted.
return this;
}
@@ -76,6 +79,7 @@
}
dataReceived = dr;
int cb = dr.getCB();
+ endTransferTime = System.currentTimeMillis();
switch (cb) {
case Presentation.CB_OK:
@@ -84,15 +88,18 @@
} catch (KeyCollisionException e) {
n.logger.log(this, "Abandoning after key collision: "+this,
Logger.MINOR);
- logSuccess(n);
+// logSuccess(n);
break;
} catch (IOException e) {
n.logger.log(this, "Cache failed on commit: "+this, e,
Logger.ERROR);
- logSuccess(n); // it's not the node's fault...
+// logSuccess(n); // it's not the node's fault...
break;
}
n.logger.log(this, "Data received successfully!: "+this,
Logger.MINOR);
- logSuccess(n);
+ routes.transferSucceeded(replyTime - routedTime, hopsToLive,
+ receivingData.length(),
+ endTransferTime - replyTime);
+// logSuccess(n);
if (storeData == null) {
NoStoreData nosd = new NoStoreData(this);
n.schedule(Core.hopTime(2), nosd);
@@ -107,12 +114,14 @@
case Presentation.CB_CACHE_FAILED:
n.logger.log(this, "Cache failed while receiving data!: "+this,
Logger.ERROR);
- logFailedTransfer(n); // sometimes this is the node's fault...
break;
case Presentation.CB_BAD_DATA:
n.logger.log(this, "Upstream node sent bad data!: "+this,
Logger.NORMAL);
- logFailedTransfer(n);
+ routes.transferFailed(replyTime - routedTime, hopsToLive,
+ receivingData.length(),
+ endTransferTime - replyTime);
+// logFailedTransfer(n);
// the null check is not really needed but I hate NPEs..
if (lastPeer != null)
routes.verityFailed();
@@ -125,9 +134,19 @@
break;
default:
- if (lastPeer != null)
- routes.transferFailed();
- logFailedTransfer(n); // better safe than sorry...
+ if (lastPeer != null) {
+ long length = receivingData.length();
+ length = searchKey.
+ getTransmissionLength(length,
+ searchKey.getPartSize(length));
+ routes.transferFailed(replyTime - routedTime,
+ hopsToLive, length,
+ endTransferTime - replyTime);
+ }
+ routes.transferFailed(replyTime - routedTime, hopsToLive,
+ receivingData.length(),
+ endTransferTime - replyTime);
+// logFailedTransfer(n); // better safe than sorry...
n.logger.log(this,
"Failed to receive data with CB "+
Presentation.getCBdescription(cb)
@@ -137,7 +156,7 @@
n.logger.log(this, "Transitioning to RequestDone from "+this,
Logger.DEBUG);
-
+ terminateRouting(false, true); // routing will ignore if we succeeded already
return new RequestDone(this);
}
}
Index: RequestState.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/RequestState.java,v
retrieving revision 1.11
retrieving revision 1.12
diff -u -r1.11 -r1.12
--- RequestState.java 19 Jun 2003 13:31:20 -0000 1.11
+++ RequestState.java 30 Aug 2003 23:16:53 -0000 1.12
@@ -147,8 +147,15 @@
public String toString() {
return getClass().getName()+": key="+searchKey+", hopsToLive="+
- hopsToLive+", id="+Fields.longToHex(id)+",ft="+ft+"@"+
- System.currentTimeMillis();
+ hopsToLive+", id="+Fields.longToHex(id)+", routes="+routes+
+ ",ft="+ft+"@"+System.currentTimeMillis();
+ }
+
+ public void terminateRouting(boolean success, boolean routingRelated) {
+ if(routes != null) {
+ routes.terminate(success, routingRelated);
+ // Don't null routes. We need it for the storedata.
+ }
}
}
Index: SendingReply.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/SendingReply.java,v
retrieving revision 1.19
retrieving revision 1.20
diff -u -r1.19 -r1.20
--- SendingReply.java 12 Aug 2003 02:41:57 -0000 1.19
+++ SendingReply.java 30 Aug 2003 23:16:53 -0000 1.20
@@ -2,6 +2,7 @@
import freenet.*;
import freenet.node.*;
+import freenet.node.rt.Routing;
import freenet.node.states.data.*;
import freenet.message.*;
import freenet.support.*;
@@ -48,12 +49,17 @@
// (there is still a chance of DataSource resetting)
// -1 means we don't know the rate for that ref.
n.diagnostics.occurrenceCounting("storeDataSendingReply",1);
- ft.storeData(n, n.rt.route(searchKey, false).
- getNextRoute(), -1, 0,
+ long length = sendingData.length();
+ length = searchKey.
+ getTransmissionLength(length,
+ searchKey.getPartSize(length));
+ Routing r = n.rt.route(searchKey, hopsToLive,
+ length, false, false);
+ ft.storeData(n, r.getNextRoute(), -1, 0,
new RequestSendCallback("StoreData", n, this));
+ r.terminate(false, false); // duhh :)
// We are sending from cache, no need to do pcaching
- }
- catch (CommunicationException e) {
+ } catch (CommunicationException e) {
n.logger.log(this,
"Failed to send back StoreData to peer " + e.peer+
" for "+this, e, Logger.MINOR);
@@ -117,5 +123,7 @@
n.logger.log(this, "Lost "+this, Logger.DEBUG);
Core.diagnostics.occurrenceCounting("lostRequestState", 1);
}
-
}
+
+
+
Index: TransferInsert.java
===================================================================
RCS file:
/cvsroot/freenet/freenet/src/freenet/node/states/request/TransferInsert.java,v
retrieving revision 1.10
retrieving revision 1.11
diff -u -r1.10 -r1.11
--- TransferInsert.java 1 Aug 2003 00:16:49 -0000 1.10
+++ TransferInsert.java 30 Aug 2003 23:16:53 -0000 1.11
@@ -54,6 +54,7 @@
receivingData.cancel();
sendingData.abort(Presentation.CB_ABORTED);
mq.addElement(qf);
+ // From the requester, so Routing does not need to know
return transition(new TransferInsertPending(this), true);
}
@@ -62,8 +63,7 @@
public State receivedMessage(Node n, QueryRestarted qr) throws StateException {
try {
super.receivedQueryRestarted(n, qr);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
// going to RequestDone with SendFailedException
receivingData.cancel();
sendingData.abort(Presentation.CB_ABORTED);
@@ -80,12 +80,11 @@
}
sendingData.abort(Presentation.CB_ABORTED);
try {
+ // Will cause a searchFailed() and a restart
super.receivedRequestInitiator(n, ri);
- }
- catch (EndOfRouteException e) {
+ } catch (EndOfRouteException e) {
return super.publicEndRoute(n);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
cancelNoInsert();
// this is going to RequestDone with no route found
return rae.state;
@@ -100,6 +99,8 @@
}
sendingData.abort(Presentation.CB_ABORTED);
mq.addElement(qr);
+ // QR from the node we routed to
+ // Will cause a searchFailed in TIP.rQR
return transition(new TransferInsertPending(this), true);
}
@@ -131,11 +132,11 @@
if(logDEBUG) n.logger.log(this, "Cancelled restart for "+this, Logger.DEBUG);
try {
insertReply(n);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
// send failed..
sendingData.abort(Presentation.CB_ABORTED); // QueryAborted already sent
transition(rae.state, false);
+ // Don't care for Routing
}
// keep on trucking..
return checkTransition(n);
@@ -150,7 +151,7 @@
super.receivedStoreData(n, sd);
return this;
}
-
+
public State receivedMessage(Node n, InsertRequest ir) {
super.receivedRequest(n, ir);
return this;
@@ -164,7 +165,8 @@
if(logDEBUG) n.logger.log(this, "CB_OK: committing "+
this, Logger.DEBUG);
tryCommit(n);
- logSuccess(n);
+ // We have already logged transferSucceeded
+// logSuccess(n);
if(logDEBUG) n.logger.log(this, "Going to AWSD for "+this,
new Exception("debug"), Logger.DEBUG);
State awsd;
@@ -182,7 +184,8 @@
n.logger.log(this, "dataReceived control bytes not CB_OK, "+
"going to RequestDone for "+this, Logger.DEBUG);
receivingData.cancel();
- logFailedTransfer(n);
+// logFailedTransfer(n);
+ // Routing dealt with in received(DataReceived)
return transition(new RequestDone(this), false);
}
}
@@ -197,17 +200,18 @@
n.logger.log(this, "Going to DataPending after key collision for "+this,
Logger.MINOR);
scheduleRestart(n, 0);
- logSuccess(n);
+// logSuccess(n);
transition(new DataPending(this), false);
} catch (IOException e) {
fail(n, "Cache failed");
n.logger.log(this, "Cache failed on commit for "+this, e, Logger.ERROR);
- logSuccess(n); // not node's fault
+ terminateRouting(false, false); // our fault
transition(new RequestDone(this), false);
}
}
public State receivedMessage(Node n, DataReceived dr) throws StateException {
+ // Routing does not care
if (receivingData != dr.source()) {
throw new BadStateException("Not my DataReceived: "+dr+" for "+this);
}
@@ -218,6 +222,7 @@
case Presentation.CB_OK:
n.logger.log(this, "Data received successfully! for "+this,
n.logger.MINOR);
+
break;
case Presentation.CB_CACHE_FAILED:
@@ -236,9 +241,8 @@
default:
fail(n, "You sent "+Presentation.getCBdescription(cb));
n.logger.log(this,
- "Failed to receive insert data with CB
"+Presentation.getCBdescription(cb)
- +", for "+this,
- Logger.MINOR);
+ "Failed to receive insert data with CB "+Presentation.
+ getCBdescription(cb)+", for "+this, Logger.MINOR);
}
return checkTransition(n);
@@ -250,6 +254,7 @@
if (sendingData != ds.source()) {
throw new BadStateException("Not my DataSent: "+ds);
}
+ endTransferTime = System.currentTimeMillis();
dataSent = ds;
int cb = ds.getCB();
switch (cb) {
@@ -257,15 +262,38 @@
case Presentation.CB_OK:
n.logger.log(this, "Insert transferred successfully! for "+this,
Logger.MINOR);
+ if(acceptedTime <= 0) {
+ n.logger.log(this, "acceptedTime not set yet!: "+this,
+ Logger.NORMAL);
+// terminateRouting(false, false);
+ // Could be restarted even in AWSD! Inserts are wierd...
+ break;
+ }
+ if(insertReplyTime <= 0) {
+ n.logger.log(this, "insertReplyTime not set yet!: "+this,
+ Logger.NORMAL);
+ terminateRouting(false, false);
+ break;
+ }
+ // FIXME: inserts are not the same. Search time will be longer.
+ if(logDEBUG)
+ n.logger.log(this, "insertReplyTime = "+insertReplyTime+
+ ", routedTime = "+routedTime+
+ ", endTransferTime = "+endTransferTime,
+ Logger.DEBUG);
+ routes.transferSucceeded(acceptedTime - routedTime, hopsToLive,
+ sendingData.length(),
+ endTransferTime - acceptedTime);
+ // Of course, NGR doesn't know about the time between insertReply and
Accepted
+ // FIXME! - but the whole Insert system needs rewriting at the behaviour
level
break;
// receivingData will be caught by checkTransition
case Presentation.CB_CACHE_FAILED:
n.logger.log(this, "Transfer of insert failed, cache broken! for "+this,
Logger.ERROR);
+ terminateRouting(false, false);
queryAborted(n);
- // Cache failed can be caused by a receive error
- // However for an insert, it's the same thing - the request is stuffed
break;
case Presentation.CB_SEND_CONN_DIED:
@@ -274,6 +302,9 @@
Presentation.getCBdescription(cb)+", for "+this,
Logger.MINOR);
scheduleRestart(n, 0);
+ routes.transferFailed(insertReplyTime - routedTime, hopsToLive,
+ sendingData.length(),
+ endTransferTime - insertReplyTime);
transition(new TransferInsertPending(this), false);
break; // nop
@@ -282,14 +313,20 @@
n.logger.log(this, "Send aborted: "+
Presentation.getCBdescription(cb)+", for "+this,
Logger.MINOR);
+ terminateRouting(false, false);
queryAborted(n);
+ // Not routed-to node's fault
break;
default:
n.logger.log(this,
"Failed to send insert data with CB "+
Presentation.getCBdescription(cb)+", for"+this,
- Logger.MINOR);
+ Logger.NORMAL);
+ // Precautionary principle
+ routes.transferFailed(insertReplyTime - routedTime, hopsToLive,
+ sendingData.length(),
+ endTransferTime - insertReplyTime);
queryAborted(n);
break;
}
Index: TransferInsertPending.java
===================================================================
RCS file:
/cvsroot/freenet/freenet/src/freenet/node/states/request/TransferInsertPending.java,v
retrieving revision 1.14
retrieving revision 1.15
diff -u -r1.14 -r1.15
--- TransferInsertPending.java 19 Jun 2003 13:31:20 -0000 1.14
+++ TransferInsertPending.java 30 Aug 2003 23:16:53 -0000 1.15
@@ -58,39 +58,36 @@
}
try {
super.receivedQueryRejected(n, qr);
- }
- catch (EndOfRouteException e) {
+ } catch (EndOfRouteException e) {
if(Core.logger.shouldLog(Logger.DEBUG))
Core.logger.log(this, "end of route exception "+searchKey,
Logger.DEBUG);
return endRoute(n);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
// we are either going to SendingReply or RequestDone with no route found
+ // So don't terminate()!
if(Core.logger.shouldLog(Logger.DEBUG))
- Core.logger.log(this, "request abort exception "+searchKey,
- rae, Logger.DEBUG);
+ Core.logger.log(this, "request abort exception "+searchKey,
+ rae, Logger.DEBUG);
if (receivingData.result() == -1)
receivingData.cancel();
return rae.state;
} finally {
cleanDoc(); // if nobody is reading, nobody will
}
-
+
return new TransferInsertPending(this);
}
-
+
public State receivedMessage(Node n, RequestInitiator ri) throws StateException {
if (this.ri == null || this.ri != ri) {
throw new BadStateException("Not my request initiator: "+ri);
}
try {
super.receivedRequestInitiator(n, ri);
- }
- catch (EndOfRouteException e) {
+ } catch (EndOfRouteException e) {
return endRoute(n);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
// this is going to RequestDone with no route found
receivingData.cancel();
return rae.state;
@@ -111,8 +108,7 @@
public State receivedMessage(Node n, QueryRestarted qr) throws StateException {
try {
super.receivedQueryRestarted(n, qr);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
// going to RequestDone with SendFailedException
receivingData.cancel();
return rae.state;
@@ -122,6 +118,7 @@
// if we didn't make it to TransferInsert yet...
public State receivedMessage(Node n, DataReceived dr) throws StateException {
+ // Routing does not care
if(logDEBUG)
n.logger.log(this, "TransferInsertPending ("+searchKey+") got "+
"DataReceived", Logger.DEBUG);
@@ -151,20 +148,23 @@
} finally {
cleanDoc(); // if nobody is reading, nobody will
}
+ terminateRouting(false, false);
return new RequestDone(this);
}
return this; // still waiting for Accepted
}
public State receivedMessage(Node n, QueryAborted qf) throws StateException {
+ // Routing does not care
if (!fromOrigPeer(qf)) {
throw new BadStateException("QueryAborted from the wrong peer!");
}
- logFailure(n);
+// logFailure(n);
queryAborted(n, qf);
receivingData.cancel();
cleanDoc(); // if nobody is reading, nobody will
+ terminateRouting(false, false);
return new RequestDone(this);
}
@@ -172,6 +172,7 @@
State ret;
try {
ReceiveData recv = this.receivingData;
+ // Will note reply time if successful
ret = super.receivedDataReply(n, dr); // could BSE out
recv.cancel(); // terminating insert chain
} finally {
@@ -179,39 +180,40 @@
}
return ret;
}
-
+
public State receivedMessage(Node n, Accepted a) throws StateException {
super.receivedAccepted(n, a);
try {
- n.logger.log(this,
- "Got Accepted " + ((System.currentTimeMillis() - dimRecvTime) / 1000)
- +"s after DataInsert", Logger.DEBUG);
+ n.logger.log(this, "Got Accepted " +
+ ((System.currentTimeMillis() - dimRecvTime) / 1000)
+ +"s after DataInsert", Logger.DEBUG);
if (doc == null)
relayInsert(n);
else {
relayInsert(n, doc);
doc = null; // Don't close it, don't hold on to it
}
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
// immediate restart, or failure
return rae.state;
}
return new TransferInsert(this);
}
-
+
public State receivedMessage(Node n, InsertReply ir) throws StateException {
if (!fromLastPeer(ir)) {
throw new BadStateException("InsertReply from wrong peer!");
}
+ insertReplyTime = System.currentTimeMillis();
if (!approved) {
cancelRestart();
accepted = true;
+ acceptedTime = insertReplyTime;
approved = true;
try {
- n.logger.log(this,
- "Got InsertReply " + ((System.currentTimeMillis() - dimRecvTime)
/ 1000)
- +"s after DataInsert", Logger.DEBUG);
+ n.logger.log(this, "Got InsertReply " +
+ ((System.currentTimeMillis() - dimRecvTime) / 1000)
+ +"s after DataInsert", Logger.DEBUG);
insertReply(n);
if (doc == null)
relayInsert(n);
@@ -219,20 +221,18 @@
relayInsert(n, doc);
doc = null; // Don't close it, don't hold on to it
}
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
return rae.state;
}
}
return new TransferInsert(this);
}
-
+
private final State endRoute(Node n) throws StateTransition {
- logFailure(n);
+// logFailure(n);
try {
insertReply(n);
- }
- catch (RequestAbortException rae) {
+ } catch (RequestAbortException rae) {
return rae.state;
}
State s = new ReceivingInsert(this);
@@ -241,7 +241,7 @@
else
return s;
}
-
+
// close the doc if necessary before leaving state.
private final void cleanDoc() {
if(logDEBUG) {
Index: TransferReply.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/TransferReply.java,v
retrieving revision 1.13
retrieving revision 1.14
diff -u -r1.13 -r1.14
--- TransferReply.java 1 Aug 2003 00:22:56 -0000 1.13
+++ TransferReply.java 30 Aug 2003 23:16:53 -0000 1.14
@@ -71,11 +71,13 @@
mq.addElement(qr);
return transition(new DataPending(this), true);
// The Pending.receivedQueryRejected will log the ngrouting time stats
+ // And tell Routing
}
// could belong after a restart
public State receivedMessage(Node n, DataNotFound dnf) {
mq.addElement(dnf);
+ // Will be handled later, including Routing
return this;
}
@@ -110,25 +112,27 @@
this, Logger.DEBUG);
try {
receivingData.commit(); // make the key available
- }
- catch (KeyCollisionException e) {
+ } catch (KeyCollisionException e) {
// this is a little bit of a hack. we jump into a
// DataPending state and then handle a restart which
// makes us check for the data in the store again
n.logger.log(this, "Going to DataPending after key collision for
"+this,
Logger.MINOR);
- logSuccess(n);
+// logSuccess(n);
+// terminateRouting(true, false);
+ // Will be terminated or restarted in the DataPending
scheduleRestart(n, 0);
transition(new DataPending(this), false);
} catch (IOException e) {
fail(n, "Cache failed");
n.logger.log(this, "Cache failed on commit for "+this, e,
Logger.ERROR);
- logSuccess(n);
+// logSuccess(n);
+ // Routing already terminated
transition(new RequestDone(this), false);
}
- logSuccess(n);
+// logSuccess(n);
if(logDEBUG) n.logger.log(this, "Going to AWSD for "+this,
new Exception("debug"), Logger.DEBUG);
State awsd;
@@ -146,7 +150,8 @@
n.logger.log(this, "dataReceived control bytes not CB_OK, "+
"going to RequestDone for "+this, Logger.DEBUG);
receivingData.cancel();
- logFailedTransfer(n);
+ terminateRouting(false, false);
+// logFailedTransfer(n);
return transition(new RequestDone(this), false);
}
}
@@ -157,21 +162,30 @@
}
dataReceived = dr;
int cb = dr.getCB();
+ endTransferTime = System.currentTimeMillis();
switch (cb) {
case Presentation.CB_OK:
n.logger.log(this, "Data received successfully! for "+this,
Logger.MINOR);
+ if(replyTime <= 0)
+ n.logger.log(this, "replyTime = "+replyTime+" for "+this,
+ Logger.NORMAL);
+ routes.transferSucceeded(replyTime - routedTime, hopsToLive,
+ receivingData.length(),
+ endTransferTime - replyTime);
break;
-
+
case Presentation.CB_CACHE_FAILED:
n.logger.log(this, "Cache failed while receiving data! for "+this,
Logger.ERROR);
+ terminateRouting(false, false);
// the repercussions will strike in the DataSent..
+ // Ignore for NGRouting, our fault
break;
-
- case Presentation.CB_BAD_DATA:
- n.logger.log(this, "Upstream node sent bad data! for "+this,
+
+ case Presentation.CB_BAD_DATA:
+ n.logger.log(this, "Upstream node sent bad data! for "+this,
Logger.NORMAL);
// the null check is not really needed but I hate NPEs..
if (lastPeer != null)
@@ -179,11 +193,14 @@
// do the restart with the DataSent
sendingData.abort(Presentation.CB_RESTARTED);
break;
-
+
case Presentation.CB_RECV_CONN_DIED:
n.logger.log(this, "Upstream node connection died for "+
this, Logger.NORMAL);
- sendingData.abort(Presentation.CB_RESTARTED);
+ routes.transferFailed(replyTime - routedTime, hopsToLive,
+ receivingData.length(),
+ endTransferTime - replyTime);
+ sendingData.abort(Presentation.CB_RESTARTED); // restart
break;
case Presentation.CB_RESTARTED:
@@ -194,8 +211,15 @@
break;
default:
- if (lastPeer != null)
- routes.transferFailed();
+ if (lastPeer != null) {
+ long length = receivingData.length();
+ length = searchKey.
+ getTransmissionLength(length,
+ searchKey.getPartSize(length));
+ routes.transferFailed(replyTime - routedTime,
+ hopsToLive, length,
+ endTransferTime - replyTime);
+ }
n.logger.log(this,
"Failed to receive data with CB "+
Presentation.getCBdescription(cb)+", for "+this,
@@ -207,6 +231,7 @@
}
public State receivedMessage(Node n, DataSent ds) throws StateException {
+ // Routing doesn't care
if (sendingData != ds.source()) {
throw new BadStateException("Not my DataSent: "+ds+" for "+this);
}
_______________________________________________
cvs mailing list
[EMAIL PROTECTED]
http://dodo.freenetproject.org/cgi-bin/mailman/listinfo/cvs
