Update of /cvsroot/freenet/freenet/src/freenet/node/states/request
In directory sc8-pr-cvs1:/tmp/cvs-serv20740/src/freenet/node/states/request
Modified Files:
AwaitingInsert.java DataPending.java FeedbackToken.java
InsertPending.java Pending.java RequestDone.java
RequestState.java SendFinished.java TransferInsert.java
TransferInsertPending.java TransferReply.java
Log Message:
6169:
Send DataRequests asynchronously too. We get a SendFinished back. If it's unsuccessful
we route to the next node... Minor fixes to SendFinished handling in different states.
If we have more than 2 connections, and one of them is not sending a trailer, queue
messages to that conn despite it being over the limits. Should limit connection open
floods... at the expense of monstrous message send times.
Make sending QueryRejected's asynchronous (relatively obscure cases, the majority are
already covered).
Increase messageStoreSize to 10,000 because I was getting lost non-RequestDone
messages.
Some more work on routing termination.
Send more DataNotFound's async.
Major bugfixes in OCM.createConn.
Don't set priority of WSL/RSL to max. Looked like it might be starving stuff.
Ignore SendFinished in RequestDone.
Logging, indenting, style, etc.
Index: AwaitingInsert.java
===================================================================
RCS file:
/cvsroot/freenet/freenet/src/freenet/node/states/request/AwaitingInsert.java,v
retrieving revision 1.7
retrieving revision 1.8
diff -u -r1.7 -r1.8
--- AwaitingInsert.java 30 Aug 2003 23:16:53 -0000 1.7
+++ AwaitingInsert.java 4 Sep 2003 22:45:06 -0000 1.8
@@ -53,6 +53,7 @@
fail(n, "I/O error receiving insert");
n.logger.log(this, "Failed to cache insert on chain "
+ Long.toHexString(id), e, Logger.ERROR);
+ terminateRouting(false, false);
return new RequestDone(this);
}
receivingData.schedule(n);
@@ -66,12 +67,14 @@
n.logger.log(this, "Did not receive expected DataInsert on chain "
+ Long.toHexString(id), Logger.MINOR);
fail(n, "DataInsert never received");
+ terminateRouting(false, false);
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();
+ terminateRouting(false, false);
return new RequestDone(this);
}
}
Index: DataPending.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/DataPending.java,v
retrieving revision 1.17
retrieving revision 1.18
diff -u -r1.17 -r1.18
--- DataPending.java 3 Sep 2003 18:10:13 -0000 1.17
+++ DataPending.java 4 Sep 2003 22:45:06 -0000 1.18
@@ -47,7 +47,7 @@
checkFailureTable(n);
super.receivedQueryRejected(n, qr);
} catch (EndOfRouteException e) {
- dataNotFound(n);
+ dataNotFound(n, System.currentTimeMillis(), true);
terminateRouting(false, true); // because routing produced the QR
return new RequestDone(this);
} catch (RequestAbortException rae) {
@@ -68,7 +68,7 @@
checkFailureTable(n);
super.receivedRequestInitiator(n, ri);
} catch (EndOfRouteException e) {
- dataNotFound(n);
+ dataNotFound(n, System.currentTimeMillis(), true);
terminateRouting(false, true);
return new RequestDone(this);
} catch (RequestAbortException rae) {
@@ -138,6 +138,12 @@
super.receivedSendFinished(n, sf);
} catch (RequestAbortException rae) {
return rae.state;
+ } catch (EndOfRouteException e) {
+ dataNotFound(n, System.currentTimeMillis(), true);
+ // We don't care if it works
+
+ terminateRouting(false, true);
+ return new RequestDone(this);
}
return this;
}
Index: FeedbackToken.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/FeedbackToken.java,v
retrieving revision 1.7
retrieving revision 1.8
diff -u -r1.7 -r1.8
--- FeedbackToken.java 3 Sep 2003 18:10:13 -0000 1.7
+++ FeedbackToken.java 4 Sep 2003 22:45:06 -0000 1.8
@@ -19,9 +19,10 @@
* @param rejected total number of QueryRejected
*/
void queryRejected(Node n, int htl, String reason, FieldSet fs,
- int unreachable, int restarted, int rejected)
+ int unreachable, int restarted, int rejected,
+ MessageSendCallback cb)
throws CommunicationException;
-
+
/**
* @param millis the maximum time until the next callback
*/
Index: InsertPending.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/InsertPending.java,v
retrieving revision 1.19
retrieving revision 1.20
diff -u -r1.19 -r1.20
--- InsertPending.java 3 Sep 2003 18:10:13 -0000 1.19
+++ InsertPending.java 4 Sep 2003 22:45:06 -0000 1.20
@@ -68,8 +68,7 @@
try {
super.receivedQueryRejected(n, qr);
} 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);
+ // Could possibly restart via AwaitingInsert
return endRoute(n);
} catch (RequestAbortException rae) {
cancelNoInsert();
@@ -271,6 +270,8 @@
super.receivedSendFinished(n, sf);
} catch (RequestAbortException rae) {
return rae.state;
+ } catch (EndOfRouteException e) {
+ return endRoute(n);
}
return this;
}
Index: Pending.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/Pending.java,v
retrieving revision 1.52
retrieving revision 1.53
diff -u -r1.52 -r1.53
--- Pending.java 3 Sep 2003 18:10:13 -0000 1.52
+++ Pending.java 4 Sep 2003 22:45:06 -0000 1.53
@@ -58,6 +58,7 @@
long searchDataRoutingTime = -1;
SendFinished feedbackSender = null;
+ SendFinished outwardSender = null;
public String toString() {
return super.toString()+", routedTime="+routedTime+", replyTime="+replyTime;
@@ -582,7 +583,8 @@
routes.routeConnected(ch.isCached());
if(logDEBUG) Core.logger.log(this, "Sending message "+r+" on "
+ch+" for "+this, Logger.DEBUG);
- ch.sendMessage(r);
+ outwardSender = new SendFinished(n, id, r.toString());
+ ch.sendMessageAsync(r, outwardSender);
if(logDEBUG) Core.logger.log(this, "Sent message "+r+" on "+ch
+" for "+this, Logger.DEBUG);
break;
@@ -597,6 +599,7 @@
if (e instanceof AuthenticationFailedException) {
routes.authFailed();
} else {
+ // Conn is dead
routes.connectFailed();
}
}
@@ -612,11 +615,14 @@
n.outboundRequestLimit.inc();
lastPeer = addr;
- scheduleRestart(n, Core.hopTime(1)); // timeout to get the Accepted
+// scheduleRestart(n, Core.hopTime(1)); // timeout to get the Accepted
+ // No timeout for the actual send
}
public void receivedSendFinished(Node n, SendFinished sf)
- throws BadStateException, RequestAbortException {
+ throws BadStateException, RequestAbortException, EndOfRouteException {
+ if(!sf.finished())
+ throw new BadStateException("Got a SendFinished that wasn't finished:
"+sf);
if(sf == feedbackSender) {
n.logger.log(this, "Got a SendFinished "+sf+" for "+this,
Logger.NORMAL /* FIXME! */ );
@@ -624,8 +630,6 @@
// Our feedback sender
// If success, cool
// If failure, throw RAE to RequestDone
- if(!sf.finished())
- throw new BadStateException("Got a SendFinished that wasn't finished:
"+sf);
if(!sf.getSuccess()) {
Exception e = sf.failCause();
if(e instanceof CommunicationException) {
@@ -641,13 +645,51 @@
terminateRouting(false, false);
throw new RequestAbortException(new RequestDone(this));
}
+ } else if(sf == outwardSender) {
+ // Our DataRequest send
+ if(!sf.getSuccess()) {
+ Exception e = sf.failCause();
+ if(e instanceof CommunicationException) {
+ ++unreachable;
+ // don't care if it's terminal or nonterminal
+ // because routing is too time-critical
+ CommunicationException ce = (CommunicationException)e;
+ if(logDEBUG)
+ n.logger.log(this, "Routing ("+this+") failure to: "
+ +ce.peerAddress() + " -- " + ce, ce,
+ Logger.DEBUG);
+ if (e instanceof AuthenticationFailedException) {
+ routes.authFailed();
+ } else {
+ // Conn is dead
+ routes.connectFailed();
+ }
+ } else {
+ n.logger.log(this, "DataRequest send caught "+e, e,
+ Logger.NORMAL);
+ // Don't inform routes, we have no idea what it is
+ }
+ // Either way, restart
+ cancelRestart();
+ searchData(n);
+ // Copied from receivedQueryRejected
+ long preGotRouteTime = System.currentTimeMillis();
+ Request newReq = createRequest(null, n.myRef);
+ gotRouteTime = System.currentTimeMillis();
+ if(logDEBUG)
+ n.logger.log(this, "Got request in
"+(gotRouteTime-preGotRouteTime)+
+ " for "+this, Logger.DEBUG);
+ sendOn(n, newReq, false);
+ } else {
+ scheduleRestart(n, Core.hopTime(1)); // timeout to get Accepted
+ }
}
}
private final void relayRestarted(Node n, long timeout,
boolean sendAsync) throws RequestAbortException {
if(sendAsync)
- feedbackSender = new SendFinished(n, id);
+ feedbackSender = new SendFinished(n, id, "Restarted");
try {
ft.restarted(n, timeout,
sendAsync ? feedbackSender : null);
Index: RequestDone.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/RequestDone.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- RequestDone.java 13 Jul 2002 19:43:15 -0000 1.2
+++ RequestDone.java 4 Sep 2003 22:45:06 -0000 1.3
@@ -14,6 +14,7 @@
/** this is in case we get a DataInsert we have to eat */
final Peer origPeer;
+ boolean logDEBUG;
public RequestDone(RequestState ancestor) {
super(ancestor.id());
@@ -31,21 +32,24 @@
// DataReply..
public State received(Node n, MessageObject mo) throws BadStateException {
+ logDEBUG = n.logger.shouldLog(Logger.DEBUG);
if (mo instanceof Accepted) {
return this;
- }
- else if (mo instanceof DataInsert
+ } else if (mo instanceof DataInsert
&& (origPeer == null
|| origPeer.equalsIdent(((DataInsert) mo).peerIdentity()))) {
- n.logger.log(this, "Eating DataInsert during RequestDone", Logger.DEBUG);
+ if(logDEBUG)
+ n.logger.log(this, "Eating DataInsert during RequestDone",
Logger.DEBUG);
((DataInsert) mo).eatData(n);
return this;
- }
- else if (!(mo instanceof Request)) {
+ } else if (mo instanceof SendFinished) {
+ if(logDEBUG)
+ n.logger.log(this, "Received "+mo+" during RequestDone", Logger.DEBUG);
+ return this;
+ } else if (!(mo instanceof Request)) {
throw new BadStateException("Received " + mo + " for Request " +
"that was already handled");
- }
- else { // A looped request
+ } else { // A looped request
try {
Request r = (Request) mo;
if (r.getSource() != null) {
@@ -54,8 +58,7 @@
r.getSource(), 0
);
}
- }
- catch (CommunicationException e) {
+ } catch (CommunicationException e) {
n.logger.log(this,
"Failed to backtrack on chain " + Long.toHexString(id),
e, Logger.DEBUG);
Index: RequestState.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/RequestState.java,v
retrieving revision 1.13
retrieving revision 1.14
diff -u -r1.13 -r1.14
--- RequestState.java 3 Sep 2003 18:10:13 -0000 1.13
+++ RequestState.java 4 Sep 2003 22:45:06 -0000 1.14
@@ -138,10 +138,12 @@
try {
if (reason == null)
reason = "(no reason given)";
+ MessageSendCallback cb =
+ new RequestSendCallback("QueryRejected ("+reason+")",
+ n, this);
ft.queryRejected(n, hopsToLive, reason, otherFields,
- unreachable, restarted, rejected);
- }
- catch (CommunicationException e) {
+ unreachable, restarted, rejected, cb);
+ } catch (CommunicationException e) {
if(logDEBUG)
n.logger.log(this, "I couldn't even fail right :-(",
e, Logger.DEBUG);
Index: SendFinished.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/SendFinished.java,v
retrieving revision 1.1
retrieving revision 1.2
diff -u -r1.1 -r1.2
--- SendFinished.java 3 Sep 2003 18:10:13 -0000 1.1
+++ SendFinished.java 4 Sep 2003 22:45:06 -0000 1.2
@@ -16,19 +16,21 @@
private Exception finishException;
private boolean succeeded;
final Node n;
+ final String logMessage;
- public SendFinished(Node n, long id) {
+ public SendFinished(Node n, long id, String logMessage) {
super(id, true);
initTime = System.currentTimeMillis();
finishTime = -1;
finishException = null;
succeeded = false;
this.n = n;
+ this.logMessage = logMessage;
}
public String toString() {
return getClass().getName()+"@ "+finishTime+":"+initTime+":"+succeeded+":"+
- finishException;
+ finishException+":"+logMessage;
}
public long startTime() {
Index: TransferInsert.java
===================================================================
RCS file:
/cvsroot/freenet/freenet/src/freenet/node/states/request/TransferInsert.java,v
retrieving revision 1.13
retrieving revision 1.14
diff -u -r1.13 -r1.14
--- TransferInsert.java 3 Sep 2003 18:10:13 -0000 1.13
+++ TransferInsert.java 4 Sep 2003 22:45:06 -0000 1.14
@@ -74,15 +74,9 @@
}
public State receivedMessage(Node n, SendFinished sf) throws StateException {
- try {
- super.receivedSendFinished(n, sf);
- } catch (RequestAbortException rae) {
- // going to RequestDone with SendFailedException
- receivingData.cancel();
- sendingData.abort(Presentation.CB_ABORTED);
- queryAborted(n);
- return transition(rae.state, false); // drop queue
- }
+ // Ignore it, we have incoming data
+ if(logDEBUG)
+ n.logger.log(this, "Ignoring "+sf+" in "+this, Logger.DEBUG);
return this;
}
Index: TransferInsertPending.java
===================================================================
RCS file:
/cvsroot/freenet/freenet/src/freenet/node/states/request/TransferInsertPending.java,v
retrieving revision 1.16
retrieving revision 1.17
diff -u -r1.16 -r1.17
--- TransferInsertPending.java 3 Sep 2003 18:10:13 -0000 1.16
+++ TransferInsertPending.java 4 Sep 2003 22:45:06 -0000 1.17
@@ -117,12 +117,20 @@
}
public State receivedMessage(Node n, SendFinished sf) throws StateException {
+ // If it's feedbackSender, we may want to terminate
+ // If it's outwardSender, we may want to go to SendingReply
try {
super.receivedSendFinished(n, sf);
} catch (RequestAbortException rae) {
// going to RequestDone with SendFailedException
+ // Lost contact with origin node
receivingData.cancel();
return rae.state;
+ } catch (EndOfRouteException e) {
+ if(Core.logger.shouldLog(Logger.DEBUG))
+ Core.logger.log(this, "end of route exception "+searchKey,
+ Logger.DEBUG);
+ return endRoute(n);
}
return this;
}
Index: TransferReply.java
===================================================================
RCS file: /cvsroot/freenet/freenet/src/freenet/node/states/request/TransferReply.java,v
retrieving revision 1.15
retrieving revision 1.16
diff -u -r1.15 -r1.16
--- TransferReply.java 3 Sep 2003 18:10:13 -0000 1.15
+++ TransferReply.java 4 Sep 2003 22:45:06 -0000 1.16
@@ -64,6 +64,7 @@
}
public State receivedMessage(Node n, SendFinished sf) throws StateException {
+ // Doesn't matter here
mq.addElement(sf);
return this;
}
_______________________________________________
cvs mailing list
[EMAIL PROTECTED]
http://dodo.freenetproject.org/cgi-bin/mailman/listinfo/cvs
