I2P_Developers/i2p.android.base
3
0
Fork 2
Code Issues 39 Pull Requests Actions 1 Packages Projects Releases Wiki Activity

2005-07-22 jrandom

Browse Source 
* Use the small thread pool for I2PTunnelHTTPServer (already used for
      I2PTunnelServer)
    * Minor memory churn reduction in I2CP
    * Small stats update
This commit is contained in:
jrandom
2005-07-23 00:15:56 +00:00
committed by zzz
parent 45767360ab
commit aeb8f02269
27 changed files with 367 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnel.java
View File

@ -1147,6 +1147,8 @@ public class I2PTunnel implements Logging, EventDispatcher {
private String getPrefix() { return '[' + _tunnelId + "]: "; } private String getPrefix() { return '[' + _tunnelId + "]: "; }
public I2PAppContext getContext() { return _context; }
/** /**
* Call this whenever we lose touch with the router involuntarily (aka the router * Call this whenever we lose touch with the router involuntarily (aka the router
* is off / crashed / etc) * is off / crashed / etc)

61
apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnelHTTPServer.java
View File

@ -35,78 +35,60 @@ public class I2PTunnelHTTPServer extends I2PTunnelServer {
public I2PTunnelHTTPServer(InetAddress host, int port, String privData, String spoofHost, Logging l, EventDispatcher notifyThis, I2PTunnel tunnel) { public I2PTunnelHTTPServer(InetAddress host, int port, String privData, String spoofHost, Logging l, EventDispatcher notifyThis, I2PTunnel tunnel) {
super(host, port, privData, l, notifyThis, tunnel); super(host, port, privData, l, notifyThis, tunnel);
_spoofHost = spoofHost; _spoofHost = spoofHost;
getTunnel().getContext().statManager().createRateStat("i2ptunnel.httpserver.blockingHandleTime", "how long the blocking handle takes to complete", "I2PTunnel.HTTPServer", new long[] { 60*1000, 10*60*1000, 3*60*60*1000 });
} }
public I2PTunnelHTTPServer(InetAddress host, int port, File privkey, String privkeyname, String spoofHost, Logging l, EventDispatcher notifyThis, I2PTunnel tunnel) { public I2PTunnelHTTPServer(InetAddress host, int port, File privkey, String privkeyname, String spoofHost, Logging l, EventDispatcher notifyThis, I2PTunnel tunnel) {
super(host, port, privkey, privkeyname, l, notifyThis, tunnel); super(host, port, privkey, privkeyname, l, notifyThis, tunnel);
_spoofHost = spoofHost; _spoofHost = spoofHost;
getTunnel().getContext().statManager().createRateStat("i2ptunnel.httpserver.blockingHandleTime", "how long the blocking handle takes to complete", "I2PTunnel.HTTPServer", new long[] { 60*1000, 10*60*1000, 3*60*60*1000 });
} }
public I2PTunnelHTTPServer(InetAddress host, int port, InputStream privData, String privkeyname, String spoofHost, Logging l, EventDispatcher notifyThis, I2PTunnel tunnel) { public I2PTunnelHTTPServer(InetAddress host, int port, InputStream privData, String privkeyname, String spoofHost, Logging l, EventDispatcher notifyThis, I2PTunnel tunnel) {
super(host, port, privData, privkeyname, l, notifyThis, tunnel); super(host, port, privData, privkeyname, l, notifyThis, tunnel);
_spoofHost = spoofHost; _spoofHost = spoofHost;
} getTunnel().getContext().statManager().createRateStat("i2ptunnel.httpserver.blockingHandleTime", "how long the blocking handle takes to complete", "I2PTunnel.HTTPServer", new long[] { 60*1000, 10*60*1000, 3*60*60*1000 });
public void run() {
try {
I2PServerSocket i2pss = sockMgr.getServerSocket();
while (true) {
I2PSocket i2ps = i2pss.accept();
if (i2ps == null) throw new I2PException("I2PServerSocket closed");
I2PThread t = new I2PThread(new Handler(i2ps));
t.start();
}
} catch (I2PException ex) {
_log.error("Error while waiting for I2PConnections", ex);
} catch (IOException ex) {
_log.error("Error while waiting for I2PConnections", ex);
}
} }
/** /**
* Async handler to keep .accept() from blocking too long. * Called by the thread pool of I2PSocket handlers
* todo: replace with a thread pool so we dont get overrun by threads if/when
* receiving a lot of connection requests concurrently.
* *
*/ */
private class Handler implements Runnable { protected void blockingHandle(I2PSocket socket) {
private I2PSocket _handleSocket; long afterAccept = getTunnel().getContext().clock().now();
public Handler(I2PSocket socket) {
_handleSocket = socket;
}
public void run() {
long afterAccept = I2PAppContext.getGlobalContext().clock().now();
long afterSocket = -1; long afterSocket = -1;
//local is fast, so synchronously. Does not need that many //local is fast, so synchronously. Does not need that many
//threads. //threads.
try { try {
_handleSocket.setReadTimeout(readTimeout); socket.setReadTimeout(readTimeout);
String modifiedHeader = getModifiedHeader(); String modifiedHeader = getModifiedHeader(socket);
if (_log.shouldLog(Log.DEBUG)) if (_log.shouldLog(Log.DEBUG))
_log.debug("Modified header: [" + modifiedHeader + "]"); _log.debug("Modified header: [" + modifiedHeader + "]");
Socket s = new Socket(remoteHost, remotePort); Socket s = new Socket(remoteHost, remotePort);
afterSocket = I2PAppContext.getGlobalContext().clock().now(); afterSocket = getTunnel().getContext().clock().now();
new I2PTunnelRunner(s, _handleSocket, slock, null, modifiedHeader.getBytes(), null); new I2PTunnelRunner(s, socket, slock, null, modifiedHeader.getBytes(), null);
} catch (SocketException ex) { } catch (SocketException ex) {
try { try {
_handleSocket.close(); socket.close();
} catch (IOException ioe) { } catch (IOException ioe) {
if (_log.shouldLog(Log.ERROR))
_log.error("Error while closing the received i2p con", ex); _log.error("Error while closing the received i2p con", ex);
} }
} catch (IOException ex) { } catch (IOException ex) {
_log.error("Error while handling for I2PConnections", ex); if (_log.shouldLog(Log.WARN))
_log.warn("Error while receiving the new HTTP request", ex);
} }
long afterHandle = I2PAppContext.getGlobalContext().clock().now(); long afterHandle = getTunnel().getContext().clock().now();
long timeToHandle = afterHandle - afterAccept; long timeToHandle = afterHandle - afterAccept;
if (timeToHandle > 1000) getTunnel().getContext().statManager().addRateData("i2ptunnel.httpserver.blockingHandleTime", timeToHandle, 0);
_log.warn("Took a while to handle the request [" + timeToHandle + ", socket create: " if ( (timeToHandle > 1000) && (_log.shouldLog(Log.WARN)) )
+ (afterSocket-afterAccept) + "]"); _log.warn("Took a while to handle the request [" + timeToHandle + ", socket create: " + (afterSocket-afterAccept) + "]");
} }
private String getModifiedHeader() throws IOException {
InputStream in = _handleSocket.getInputStream(); private String getModifiedHeader(I2PSocket handleSocket) throws IOException {
InputStream in = handleSocket.getInputStream();
StringBuffer command = new StringBuffer(128); StringBuffer command = new StringBuffer(128);
Properties headers = readHeaders(in, command); Properties headers = readHeaders(in, command);
@ -114,7 +96,6 @@ public class I2PTunnelHTTPServer extends I2PTunnelServer {
headers.setProperty("Connection", "close"); headers.setProperty("Connection", "close");
return formatHeaders(headers, command); return formatHeaders(headers, command);
} }
}
private String formatHeaders(Properties headers, StringBuffer command) { private String formatHeaders(Properties headers, StringBuffer command) {
StringBuffer buf = new StringBuffer(command.length() + headers.size() * 64); StringBuffer buf = new StringBuffer(command.length() + headers.size() * 64);

6
apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnelServer.java
View File

@ -189,7 +189,7 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
public void run() { public void run() {
while (open) { while (open) {
try { try {
handle(_serverSocket.accept()); blockingHandle(_serverSocket.accept());
} catch (I2PException ex) { } catch (I2PException ex) {
_log.error("Error while waiting for I2PConnections", ex); _log.error("Error while waiting for I2PConnections", ex);
return; return;
@ -199,8 +199,9 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
} }
} }
} }
}
private void handle(I2PSocket socket) { protected void blockingHandle(I2PSocket socket) {
long afterAccept = I2PAppContext.getGlobalContext().clock().now(); long afterAccept = I2PAppContext.getGlobalContext().clock().now();
long afterSocket = -1; long afterSocket = -1;
//local is fast, so synchronously. Does not need that many //local is fast, so synchronously. Does not need that many
@ -225,6 +226,5 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
if (timeToHandle > 1000) if (timeToHandle > 1000)
_log.warn("Took a while to handle the request [" + timeToHandle + ", socket create: " + (afterSocket-afterAccept) + "]"); _log.warn("Took a while to handle the request [" + timeToHandle + ", socket create: " + (afterSocket-afterAccept) + "]");
} }
}
} }

4
apps/streaming/java/src/net/i2p/client/streaming/MessageHandler.java
View File

@ -91,8 +91,8 @@ public class MessageHandler implements I2PSessionListener {
* *
*/ */
public void errorOccurred(I2PSession session, String message, Throwable error) { public void errorOccurred(I2PSession session, String message, Throwable error) {
if (_log.shouldLog(Log.ERROR)) if (_log.shouldLog(Log.WARN))
_log.error("error occurred: " + message + "- " + error.getMessage()); _log.warn("error occurred: " + message + "- " + error.getMessage());
if (_log.shouldLog(Log.WARN)) if (_log.shouldLog(Log.WARN))
_log.warn("cause", error); _log.warn("cause", error);
//_manager.disconnectAllHard(); //_manager.disconnectAllHard();

27
core/java/src/net/i2p/client/I2PClientMessageHandlerMap.java
View File

@ -29,20 +29,27 @@ import net.i2p.util.Log;
class I2PClientMessageHandlerMap { class I2PClientMessageHandlerMap {
private final static Log _log = new Log(I2PClientMessageHandlerMap.class); private final static Log _log = new Log(I2PClientMessageHandlerMap.class);
/** map of message type id --> I2CPMessageHandler */ /** map of message type id --> I2CPMessageHandler */
private Map _handlers; private I2CPMessageHandler _handlers[];
public I2PClientMessageHandlerMap(I2PAppContext context) { public I2PClientMessageHandlerMap(I2PAppContext context) {
_handlers = new HashMap(); int highest = DisconnectMessage.MESSAGE_TYPE;
_handlers.put(new Integer(DisconnectMessage.MESSAGE_TYPE), new DisconnectMessageHandler(context)); highest = Math.max(highest, SessionStatusMessage.MESSAGE_TYPE);
_handlers.put(new Integer(SessionStatusMessage.MESSAGE_TYPE), new SessionStatusMessageHandler(context)); highest = Math.max(highest, RequestLeaseSetMessage.MESSAGE_TYPE);
_handlers.put(new Integer(RequestLeaseSetMessage.MESSAGE_TYPE), new RequestLeaseSetMessageHandler(context)); highest = Math.max(highest, MessagePayloadMessage.MESSAGE_TYPE);
_handlers.put(new Integer(MessagePayloadMessage.MESSAGE_TYPE), new MessagePayloadMessageHandler(context)); highest = Math.max(highest, MessageStatusMessage.MESSAGE_TYPE);
_handlers.put(new Integer(MessageStatusMessage.MESSAGE_TYPE), new MessageStatusMessageHandler(context)); highest = Math.max(highest, SetDateMessage.MESSAGE_TYPE);
_handlers.put(new Integer(SetDateMessage.MESSAGE_TYPE), new SetDateMessageHandler(context));
_handlers = new I2CPMessageHandler[highest+1];
_handlers[DisconnectMessage.MESSAGE_TYPE] = new DisconnectMessageHandler(context);
_handlers[SessionStatusMessage.MESSAGE_TYPE] = new SessionStatusMessageHandler(context);
_handlers[RequestLeaseSetMessage.MESSAGE_TYPE] = new RequestLeaseSetMessageHandler(context);
_handlers[MessagePayloadMessage.MESSAGE_TYPE] = new MessagePayloadMessageHandler(context);
_handlers[MessageStatusMessage.MESSAGE_TYPE] = new MessageStatusMessageHandler(context);
_handlers[SetDateMessage.MESSAGE_TYPE] = new SetDateMessageHandler(context);
} }
public I2CPMessageHandler getHandler(int messageTypeId) { public I2CPMessageHandler getHandler(int messageTypeId) {
I2CPMessageHandler handler = (I2CPMessageHandler) _handlers.get(new Integer(messageTypeId)); if ( (messageTypeId < 0) || (messageTypeId >= _handlers.length) ) return null;
return handler; return _handlers[messageTypeId];
} }
} }

26
core/java/src/net/i2p/client/I2PSessionImpl.java
View File

@ -78,7 +78,7 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
/** class that generates new messages */ /** class that generates new messages */
protected I2CPMessageProducer _producer; protected I2CPMessageProducer _producer;
/** map of integer --> MessagePayloadMessage */ /** map of Long --> MessagePayloadMessage */
private Map _availableMessages; private Map _availableMessages;
protected I2PClientMessageHandlerMap _handlerMap; protected I2PClientMessageHandlerMap _handlerMap;
@ -295,7 +295,7 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
public byte[] receiveMessage(int msgId) throws I2PSessionException { public byte[] receiveMessage(int msgId) throws I2PSessionException {
MessagePayloadMessage msg = null; MessagePayloadMessage msg = null;
synchronized (_availableMessages) { synchronized (_availableMessages) {
msg = (MessagePayloadMessage) _availableMessages.remove(new Integer(msgId)); msg = (MessagePayloadMessage) _availableMessages.remove(new Long(msgId));
} }
if (msg == null) return null; if (msg == null) return null;
return msg.getPayload().getUnencryptedData(); return msg.getPayload().getUnencryptedData();
@ -346,9 +346,9 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
*/ */
public void addNewMessage(MessagePayloadMessage msg) { public void addNewMessage(MessagePayloadMessage msg) {
synchronized (_availableMessages) { synchronized (_availableMessages) {
_availableMessages.put(new Integer(msg.getMessageId().getMessageId()), msg); _availableMessages.put(new Long(msg.getMessageId()), msg);
} }
int id = msg.getMessageId().getMessageId(); long id = msg.getMessageId();
byte data[] = msg.getPayload().getUnencryptedData(); byte data[] = msg.getPayload().getUnencryptedData();
if ((data == null) || (data.length <= 0)) { if ((data == null) || (data.length <= 0)) {
if (_log.shouldLog(Log.CRIT)) if (_log.shouldLog(Log.CRIT))
@ -363,12 +363,12 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
SimpleTimer.getInstance().addEvent(new VerifyUsage(id), 30*1000); SimpleTimer.getInstance().addEvent(new VerifyUsage(id), 30*1000);
} }
private class VerifyUsage implements SimpleTimer.TimedEvent { private class VerifyUsage implements SimpleTimer.TimedEvent {
private int _msgId; private long _msgId;
public VerifyUsage(int id) { _msgId = id; } public VerifyUsage(long id) { _msgId = id; }
public void timeReached() { public void timeReached() {
MessagePayloadMessage removed = null; MessagePayloadMessage removed = null;
synchronized (_availableMessages) { synchronized (_availableMessages) {
removed = (MessagePayloadMessage)_availableMessages.remove(new Integer(_msgId)); removed = (MessagePayloadMessage)_availableMessages.remove(new Long(_msgId));
} }
if (removed != null) if (removed != null)
_log.log(Log.CRIT, "Message NOT removed! id=" + _msgId + ": " + removed); _log.log(Log.CRIT, "Message NOT removed! id=" + _msgId + ": " + removed);
@ -393,9 +393,9 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
} }
} }
public void available(int msgId, int size) { public void available(long msgId, int size) {
synchronized (AvailabilityNotifier.this) { synchronized (AvailabilityNotifier.this) {
_pendingIds.add(new Integer(msgId)); _pendingIds.add(new Long(msgId));
_pendingSizes.add(new Integer(size)); _pendingSizes.add(new Integer(size));
AvailabilityNotifier.this.notifyAll(); AvailabilityNotifier.this.notifyAll();
} }
@ -403,7 +403,7 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
public void run() { public void run() {
_alive = true; _alive = true;
while (_alive) { while (_alive) {
Integer msgId = null; Long msgId = null;
Integer size = null; Integer size = null;
synchronized (AvailabilityNotifier.this) { synchronized (AvailabilityNotifier.this) {
if (_pendingIds.size() <= 0) { if (_pendingIds.size() <= 0) {
@ -413,7 +413,7 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
} }
} }
if (_pendingIds.size() > 0) { if (_pendingIds.size() > 0) {
msgId = (Integer)_pendingIds.remove(0); msgId = (Long)_pendingIds.remove(0);
size = (Integer)_pendingSizes.remove(0); size = (Integer)_pendingSizes.remove(0);
} }
} }
@ -532,8 +532,8 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
* Pass off the error to the listener * Pass off the error to the listener
*/ */
void propogateError(String msg, Throwable error) { void propogateError(String msg, Throwable error) {
if (_log.shouldLog(Log.ERROR)) if (_log.shouldLog(Log.WARN))
_log.error(getPrefix() + "Error occurred: " + msg + " - " + error.getMessage()); _log.warn(getPrefix() + "Error occurred: " + msg + " - " + error.getMessage());
if (_log.shouldLog(Log.WARN)) if (_log.shouldLog(Log.WARN))
_log.warn(getPrefix() + " cause", error); _log.warn(getPrefix() + " cause", error);

2
core/java/src/net/i2p/client/MessagePayloadMessageHandler.java
View File

@ -35,7 +35,7 @@ class MessagePayloadMessageHandler extends HandlerImpl {
_log.debug("Handle message " + message); _log.debug("Handle message " + message);
try { try {
MessagePayloadMessage msg = (MessagePayloadMessage) message; MessagePayloadMessage msg = (MessagePayloadMessage) message;
MessageId id = msg.getMessageId(); long id = msg.getMessageId();
Payload payload = decryptPayload(msg, session); Payload payload = decryptPayload(msg, session);
session.addNewMessage(msg); session.addNewMessage(msg);

6
core/java/src/net/i2p/client/MessageStatusMessageHandler.java
View File

@ -46,7 +46,7 @@ class MessageStatusMessageHandler extends HandlerImpl {
} }
return; return;
case MessageStatusMessage.STATUS_SEND_ACCEPTED: case MessageStatusMessage.STATUS_SEND_ACCEPTED:
session.receiveStatus(msg.getMessageId().getMessageId(), msg.getNonce(), msg.getStatus()); session.receiveStatus((int)msg.getMessageId(), msg.getNonce(), msg.getStatus());
// noop // noop
return; return;
case MessageStatusMessage.STATUS_SEND_BEST_EFFORT_SUCCESS: case MessageStatusMessage.STATUS_SEND_BEST_EFFORT_SUCCESS:
@ -54,14 +54,14 @@ class MessageStatusMessageHandler extends HandlerImpl {
if (_log.shouldLog(Log.INFO)) if (_log.shouldLog(Log.INFO))
_log.info("Message delivery succeeded for message " + msg.getMessageId()); _log.info("Message delivery succeeded for message " + msg.getMessageId());
//if (!skipStatus) //if (!skipStatus)
session.receiveStatus(msg.getMessageId().getMessageId(), msg.getNonce(), msg.getStatus()); session.receiveStatus((int)msg.getMessageId(), msg.getNonce(), msg.getStatus());
return; return;
case MessageStatusMessage.STATUS_SEND_BEST_EFFORT_FAILURE: case MessageStatusMessage.STATUS_SEND_BEST_EFFORT_FAILURE:
case MessageStatusMessage.STATUS_SEND_GUARANTEED_FAILURE: case MessageStatusMessage.STATUS_SEND_GUARANTEED_FAILURE:
if (_log.shouldLog(Log.INFO)) if (_log.shouldLog(Log.INFO))
_log.info("Message delivery FAILED for message " + msg.getMessageId()); _log.info("Message delivery FAILED for message " + msg.getMessageId());
//if (!skipStatus) //if (!skipStatus)
session.receiveStatus(msg.getMessageId().getMessageId(), msg.getNonce(), msg.getStatus()); session.receiveStatus((int)msg.getMessageId(), msg.getNonce(), msg.getStatus());
return; return;
default: default:
if (_log.shouldLog(Log.ERROR)) if (_log.shouldLog(Log.ERROR))

3
core/java/src/net/i2p/data/i2cp/I2CPMessageImpl.java
View File

@ -61,6 +61,7 @@ public abstract class I2CPMessageImpl extends DataStructureImpl implements I2CPM
+ " class: " + getClass().getName() + ")"); + " class: " + getClass().getName() + ")");
if (length < 0) throw new IOException("Negative payload size"); if (length < 0) throw new IOException("Negative payload size");
/*
byte buf[] = new byte[length]; byte buf[] = new byte[length];
int read = DataHelper.read(in, buf); int read = DataHelper.read(in, buf);
if (read != length) if (read != length)
@ -69,6 +70,8 @@ public abstract class I2CPMessageImpl extends DataStructureImpl implements I2CPM
ByteArrayInputStream bis = new ByteArrayInputStream(buf); ByteArrayInputStream bis = new ByteArrayInputStream(buf);
doReadMessage(bis, length); doReadMessage(bis, length);
*/
doReadMessage(in, length);
} }
/** /**

13
core/java/src/net/i2p/data/i2cp/MessageId.java
View File

@ -26,22 +26,25 @@ import net.i2p.util.Log;
*/ */
public class MessageId extends DataStructureImpl { public class MessageId extends DataStructureImpl {
private final static Log _log = new Log(MessageId.class); private final static Log _log = new Log(MessageId.class);
private int _messageId; private long _messageId;
public MessageId() { public MessageId() {
setMessageId(-1); setMessageId(-1);
} }
public MessageId(long id) {
setMessageId(id);
}
public int getMessageId() { public long getMessageId() {
return _messageId; return _messageId;
} }
public void setMessageId(int id) { public void setMessageId(long id) {
_messageId = id; _messageId = id;
} }
public void readBytes(InputStream in) throws DataFormatException, IOException { public void readBytes(InputStream in) throws DataFormatException, IOException {
_messageId = (int) DataHelper.readLong(in, 4); _messageId = DataHelper.readLong(in, 4);
} }
public void writeBytes(OutputStream out) throws DataFormatException, IOException { public void writeBytes(OutputStream out) throws DataFormatException, IOException {
@ -55,7 +58,7 @@ public class MessageId extends DataStructureImpl {
} }
public int hashCode() { public int hashCode() {
return getMessageId(); return (int)getMessageId();
} }
public String toString() { public String toString() {

30
core/java/src/net/i2p/data/i2cp/MessagePayloadMessage.java
View File

@ -27,29 +27,29 @@ import net.i2p.util.Log;
public class MessagePayloadMessage extends I2CPMessageImpl { public class MessagePayloadMessage extends I2CPMessageImpl {
private final static Log _log = new Log(MessagePayloadMessage.class); private final static Log _log = new Log(MessagePayloadMessage.class);
public final static int MESSAGE_TYPE = 31; public final static int MESSAGE_TYPE = 31;
private SessionId _sessionId; private long _sessionId;
private MessageId _messageId; private long _messageId;
private Payload _payload; private Payload _payload;
public MessagePayloadMessage() { public MessagePayloadMessage() {
setSessionId(null); setSessionId(-1);
setMessageId(null); setMessageId(-1);
setPayload(null); setPayload(null);
} }
public SessionId getSessionId() { public long getSessionId() {
return _sessionId; return _sessionId;
} }
public void setSessionId(SessionId id) { public void setSessionId(long id) {
_sessionId = id; _sessionId = id;
} }
public MessageId getMessageId() { public long getMessageId() {
return _messageId; return _messageId;
} }
public void setMessageId(MessageId id) { public void setMessageId(long id) {
_messageId = id; _messageId = id;
} }
@ -63,10 +63,8 @@ public class MessagePayloadMessage extends I2CPMessageImpl {
protected void doReadMessage(InputStream in, int size) throws I2CPMessageException, IOException { protected void doReadMessage(InputStream in, int size) throws I2CPMessageException, IOException {
try { try {
_sessionId = new SessionId(); _sessionId = DataHelper.readLong(in, 2);
_sessionId.readBytes(in); _messageId = DataHelper.readLong(in, 4);
_messageId = new MessageId();
_messageId.readBytes(in);
_payload = new Payload(); _payload = new Payload();
_payload.readBytes(in); _payload.readBytes(in);
} catch (DataFormatException dfe) { } catch (DataFormatException dfe) {
@ -84,9 +82,9 @@ public class MessagePayloadMessage extends I2CPMessageImpl {
* *
*/ */
public void writeMessage(OutputStream out) throws I2CPMessageException, IOException { public void writeMessage(OutputStream out) throws I2CPMessageException, IOException {
if (_sessionId == null) if (_sessionId <= 0)
throw new I2CPMessageException("Unable to write out the message, as the session ID has not been defined"); throw new I2CPMessageException("Unable to write out the message, as the session ID has not been defined");
if (_messageId == null) if (_messageId < 0)
throw new I2CPMessageException("Unable to write out the message, as the message ID has not been defined"); throw new I2CPMessageException("Unable to write out the message, as the message ID has not been defined");
if (_payload == null) if (_payload == null)
throw new I2CPMessageException("Unable to write out the message, as the payload has not been defined"); throw new I2CPMessageException("Unable to write out the message, as the payload has not been defined");
@ -95,8 +93,8 @@ public class MessagePayloadMessage extends I2CPMessageImpl {
try { try {
DataHelper.writeLong(out, 4, size); DataHelper.writeLong(out, 4, size);
DataHelper.writeLong(out, 1, getType()); DataHelper.writeLong(out, 1, getType());
DataHelper.writeLong(out, 2, _sessionId.getSessionId()); DataHelper.writeLong(out, 2, _sessionId);
DataHelper.writeLong(out, 4, _messageId.getMessageId()); DataHelper.writeLong(out, 4, _messageId);
DataHelper.writeLong(out, 4, _payload.getSize()); DataHelper.writeLong(out, 4, _payload.getSize());
out.write(_payload.getEncryptedData()); out.write(_payload.getEncryptedData());
} catch (DataFormatException dfe) { } catch (DataFormatException dfe) {

57
core/java/src/net/i2p/data/i2cp/MessageStatusMessage.java
View File

@ -12,6 +12,7 @@ package net.i2p.data.i2cp;
import java.io.ByteArrayOutputStream; import java.io.ByteArrayOutputStream;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.io.OutputStream;
import net.i2p.data.DataFormatException; import net.i2p.data.DataFormatException;
import net.i2p.data.DataHelper; import net.i2p.data.DataHelper;
@ -26,8 +27,8 @@ import net.i2p.util.Log;
public class MessageStatusMessage extends I2CPMessageImpl { public class MessageStatusMessage extends I2CPMessageImpl {
private final static Log _log = new Log(SessionStatusMessage.class); private final static Log _log = new Log(SessionStatusMessage.class);
public final static int MESSAGE_TYPE = 22; public final static int MESSAGE_TYPE = 22;
private SessionId _sessionId; private long _sessionId;
private MessageId _messageId; private long _messageId;
private long _nonce; private long _nonce;
private long _size; private long _size;
private int _status; private int _status;
@ -40,18 +41,18 @@ public class MessageStatusMessage extends I2CPMessageImpl {
public final static int STATUS_SEND_GUARANTEED_FAILURE = 5; public final static int STATUS_SEND_GUARANTEED_FAILURE = 5;
public MessageStatusMessage() { public MessageStatusMessage() {
setSessionId(null); setSessionId(-1);
setStatus(-1); setStatus(-1);
setMessageId(null); setMessageId(-1);
setSize(-1); setSize(-1);
setNonce(-1); setNonce(-1);
} }
public SessionId getSessionId() { public long getSessionId() {
return _sessionId; return _sessionId;
} }
public void setSessionId(SessionId id) { public void setSessionId(long id) {
_sessionId = id; _sessionId = id;
} }
@ -63,11 +64,11 @@ public class MessageStatusMessage extends I2CPMessageImpl {
_status = status; _status = status;
} }
public MessageId getMessageId() { public long getMessageId() {
return _messageId; return _messageId;
} }
public void setMessageId(MessageId id) { public void setMessageId(long id) {
_messageId = id; _messageId = id;
} }
@ -108,10 +109,8 @@ public class MessageStatusMessage extends I2CPMessageImpl {
protected void doReadMessage(InputStream in, int size) throws I2CPMessageException, IOException { protected void doReadMessage(InputStream in, int size) throws I2CPMessageException, IOException {
try { try {
_sessionId = new SessionId(); _sessionId = DataHelper.readLong(in, 2);
_sessionId.readBytes(in); _messageId = DataHelper.readLong(in, 4);
_messageId = new MessageId();
_messageId.readBytes(in);
_status = (int) DataHelper.readLong(in, 1); _status = (int) DataHelper.readLong(in, 1);
_size = DataHelper.readLong(in, 4); _size = DataHelper.readLong(in, 4);
_nonce = DataHelper.readLong(in, 4); _nonce = DataHelper.readLong(in, 4);
@ -120,20 +119,32 @@ public class MessageStatusMessage extends I2CPMessageImpl {
} }
} }
protected byte[] doWriteMessage() throws I2CPMessageException, IOException {
if ((_sessionId == null) || (_messageId == null) || (_status < 0) || (_nonce <= 0)) /**
throw new I2CPMessageException("Unable to write out the message as there is not enough data"); * Override to reduce mem churn
ByteArrayOutputStream os = new ByteArrayOutputStream(64); */
public void writeMessage(OutputStream out) throws I2CPMessageException, IOException {
int len = 2 + // sessionId
4 + // messageId
1 + // status
4 + // size
4; // nonce
try { try {
_sessionId.writeBytes(os); DataHelper.writeLong(out, 4, len);
_messageId.writeBytes(os); DataHelper.writeLong(out, 1, getType());
DataHelper.writeLong(os, 1, _status); DataHelper.writeLong(out, 2, _sessionId);
DataHelper.writeLong(os, 4, _size); DataHelper.writeLong(out, 4, _messageId);
DataHelper.writeLong(os, 4, _nonce); DataHelper.writeLong(out, 1, _status);
DataHelper.writeLong(out, 4, _size);
DataHelper.writeLong(out, 4, _nonce);
} catch (DataFormatException dfe) { } catch (DataFormatException dfe) {
throw new I2CPMessageException("Error writing out the message data", dfe); throw new I2CPMessageException("Unable to write the message length or type", dfe);
} }
return os.toByteArray(); }
protected byte[] doWriteMessage() throws I2CPMessageException, IOException {
throw new UnsupportedOperationException("This shouldn't be called... use writeMessage(out)");
} }
public int getType() { public int getType() {

49
core/java/src/net/i2p/data/i2cp/ReceiveMessageBeginMessage.java
View File

@ -12,6 +12,7 @@ package net.i2p.data.i2cp;
import java.io.ByteArrayOutputStream; import java.io.ByteArrayOutputStream;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.io.OutputStream;
import net.i2p.data.DataFormatException; import net.i2p.data.DataFormatException;
import net.i2p.data.DataHelper; import net.i2p.data.DataHelper;
@ -26,50 +27,62 @@ import net.i2p.util.Log;
public class ReceiveMessageBeginMessage extends I2CPMessageImpl { public class ReceiveMessageBeginMessage extends I2CPMessageImpl {
private final static Log _log = new Log(ReceiveMessageBeginMessage.class); private final static Log _log = new Log(ReceiveMessageBeginMessage.class);
public final static int MESSAGE_TYPE = 6; public final static int MESSAGE_TYPE = 6;
private SessionId _sessionId; private long _sessionId;
private MessageId _messageId; private long _messageId;
public ReceiveMessageBeginMessage() { public ReceiveMessageBeginMessage() {
setSessionId(null); setSessionId(-1);
setMessageId(null); setMessageId(-1);
} }
public SessionId getSessionId() { public long getSessionId() {
return _sessionId; return _sessionId;
} }
public void setSessionId(SessionId id) { public void setSessionId(long id) {
_sessionId = id; _sessionId = id;
} }
public MessageId getMessageId() { public long getMessageId() {
return _messageId; return _messageId;
} }
public void setMessageId(MessageId id) { public void setMessageId(long id) {
_messageId = id; _messageId = id;
} }
protected void doReadMessage(InputStream in, int size) throws I2CPMessageException, IOException { protected void doReadMessage(InputStream in, int size) throws I2CPMessageException, IOException {
try { try {
_sessionId = new SessionId(); _sessionId = DataHelper.readLong(in, 2);
_sessionId.readBytes(in); _messageId = DataHelper.readLong(in, 4);
_messageId = new MessageId();
_messageId.readBytes(in);
} catch (DataFormatException dfe) { } catch (DataFormatException dfe) {
throw new I2CPMessageException("Unable to load the message data", dfe); throw new I2CPMessageException("Unable to load the message data", dfe);
} }
} }
protected byte[] doWriteMessage() throws I2CPMessageException, IOException { protected byte[] doWriteMessage() throws I2CPMessageException, IOException {
if ((_sessionId == null) || (_messageId == null)) throw new UnsupportedOperationException("This shouldn't be called... use writeMessage(out)");
throw new I2CPMessageException("Unable to write out the message as there is not enough data");
byte rv[] = new byte[2+4];
DataHelper.toLong(rv, 0, 2, _sessionId.getSessionId());
DataHelper.toLong(rv, 2, 4, _messageId.getMessageId());
return rv;
} }
/**
* Override to reduce mem churn
*/
public void writeMessage(OutputStream out) throws I2CPMessageException, IOException {
int len = 2 + // sessionId
4; // messageId
try {
DataHelper.writeLong(out, 4, len);
DataHelper.writeLong(out, 1, getType());
DataHelper.writeLong(out, 2, _sessionId);
DataHelper.writeLong(out, 4, _messageId);
} catch (DataFormatException dfe) {
throw new I2CPMessageException("Unable to write the message length or type", dfe);
}
}
public int getType() { public int getType() {
return MESSAGE_TYPE; return MESSAGE_TYPE;
} }

28
core/java/src/net/i2p/data/i2cp/ReceiveMessageEndMessage.java
View File

@ -26,47 +26,45 @@ import net.i2p.util.Log;
public class ReceiveMessageEndMessage extends I2CPMessageImpl { public class ReceiveMessageEndMessage extends I2CPMessageImpl {
private final static Log _log = new Log(ReceiveMessageEndMessage.class); private final static Log _log = new Log(ReceiveMessageEndMessage.class);
public final static int MESSAGE_TYPE = 7; public final static int MESSAGE_TYPE = 7;
private SessionId _sessionId; private long _sessionId;
private MessageId _messageId; private long _messageId;
public ReceiveMessageEndMessage() { public ReceiveMessageEndMessage() {
setSessionId(null); setSessionId(-1);
setMessageId(null); setMessageId(-1);
} }
public SessionId getSessionId() { public long getSessionId() {
return _sessionId; return _sessionId;
} }
public void setSessionId(SessionId id) { public void setSessionId(long id) {
_sessionId = id; _sessionId = id;
} }
public MessageId getMessageId() { public long getMessageId() {
return _messageId; return _messageId;
} }
public void setMessageId(MessageId id) { public void setMessageId(long id) {
_messageId = id; _messageId = id;
} }
protected void doReadMessage(InputStream in, int size) throws I2CPMessageException, IOException { protected void doReadMessage(InputStream in, int size) throws I2CPMessageException, IOException {
try { try {
_sessionId = new SessionId(); _sessionId = DataHelper.readLong(in, 2);
_sessionId.readBytes(in); _messageId = DataHelper.readLong(in, 4);
_messageId = new MessageId();
_messageId.readBytes(in);
} catch (DataFormatException dfe) { } catch (DataFormatException dfe) {
throw new I2CPMessageException("Unable to load the message data", dfe); throw new I2CPMessageException("Unable to load the message data", dfe);
} }
} }
protected byte[] doWriteMessage() throws I2CPMessageException, IOException { protected byte[] doWriteMessage() throws I2CPMessageException, IOException {
if ((_sessionId == null) || (_messageId == null)) if ((_sessionId < 0) || (_messageId < 0))
throw new I2CPMessageException("Unable to write out the message as there is not enough data"); throw new I2CPMessageException("Unable to write out the message as there is not enough data");
byte rv[] = new byte[2+4]; byte rv[] = new byte[2+4];
DataHelper.toLong(rv, 0, 2, _sessionId.getSessionId()); DataHelper.toLong(rv, 0, 2, _sessionId);
DataHelper.toLong(rv, 2, 4, _messageId.getMessageId()); DataHelper.toLong(rv, 2, 4, _messageId);
return rv; return rv;
} }

8
history.txt
View File

@ -1,4 +1,10 @@
$Id: history.txt,v 1.215 2005/07/20 14:24:47 jrandom Exp $ $Id: history.txt,v 1.216 2005/07/21 17:37:16 jrandom Exp $
2005-07-22 jrandom
* Use the small thread pool for I2PTunnelHTTPServer (already used for
I2PTunnelServer)
* Minor memory churn reduction in I2CP
* Small stats update
2005-07-21 jrandom 2005-07-21 jrandom
* Fix in the SDK for a bug which would manifest itself as misrouted * Fix in the SDK for a bug which would manifest itself as misrouted

58
router/java/src/net/i2p/data/i2np/DeliveryInstructions.java
View File

@ -216,16 +216,48 @@ public class DeliveryInstructions extends DataStructureImpl {
val = val | fmode; val = val | fmode;
if (getDelayRequested()) if (getDelayRequested())
val = val | FLAG_DELAY; val = val | FLAG_DELAY;
if (_log.shouldLog(Log.DEBUG))
_log.debug("getFlags() = " + val); _log.debug("getFlags() = " + val);
return val; return val;
} }
private byte[] getAdditionalInfo() throws DataFormatException { private byte[] getAdditionalInfo() throws DataFormatException {
ByteArrayOutputStream baos = new ByteArrayOutputStream(64); int additionalSize = 0;
try {
if (getEncrypted()) { if (getEncrypted()) {
if (_encryptionKey == null) throw new DataFormatException("Encryption key is not set"); if (_encryptionKey == null) throw new DataFormatException("Encryption key is not set");
_encryptionKey.writeBytes(baos); additionalSize += SessionKey.KEYSIZE_BYTES;
}
switch (getDeliveryMode()) {
case FLAG_MODE_LOCAL:
if (_log.shouldLog(Log.DEBUG))
_log.debug("mode = local");
break;
case FLAG_MODE_DESTINATION:
if (_destinationHash == null) throw new DataFormatException("Destination hash is not set");
additionalSize += Hash.HASH_LENGTH;
break;
case FLAG_MODE_ROUTER:
if (_routerHash == null) throw new DataFormatException("Router hash is not set");
additionalSize += Hash.HASH_LENGTH;
break;
case FLAG_MODE_TUNNEL:
if ( (_routerHash == null) || (_tunnelId == null) ) throw new DataFormatException("Router hash or tunnel ID is not set");
additionalSize += Hash.HASH_LENGTH;
additionalSize += 4; // tunnelId
break;
}
if (getDelayRequested()) {
additionalSize += 4;
}
byte rv[] = new byte[additionalSize];
int offset = 0;
if (getEncrypted()) {
if (_encryptionKey == null) throw new DataFormatException("Encryption key is not set");
System.arraycopy(_encryptionKey.getData(), 0, rv, offset, SessionKey.KEYSIZE_BYTES);
offset += SessionKey.KEYSIZE_BYTES;
if (_log.shouldLog(Log.DEBUG)) if (_log.shouldLog(Log.DEBUG))
_log.debug("IsEncrypted"); _log.debug("IsEncrypted");
} else { } else {
@ -239,20 +271,24 @@ public class DeliveryInstructions extends DataStructureImpl {
break; break;
case FLAG_MODE_DESTINATION: case FLAG_MODE_DESTINATION:
if (_destinationHash == null) throw new DataFormatException("Destination hash is not set"); if (_destinationHash == null) throw new DataFormatException("Destination hash is not set");
_destinationHash.writeBytes(baos); System.arraycopy(_destinationHash.getData(), 0, rv, offset, Hash.HASH_LENGTH);
offset += Hash.HASH_LENGTH;
if (_log.shouldLog(Log.DEBUG)) if (_log.shouldLog(Log.DEBUG))
_log.debug("mode = destination, hash = " + _destinationHash); _log.debug("mode = destination, hash = " + _destinationHash);
break; break;
case FLAG_MODE_ROUTER: case FLAG_MODE_ROUTER:
if (_routerHash == null) throw new DataFormatException("Router hash is not set"); if (_routerHash == null) throw new DataFormatException("Router hash is not set");
_routerHash.writeBytes(baos); System.arraycopy(_routerHash.getData(), 0, rv, offset, Hash.HASH_LENGTH);
offset += Hash.HASH_LENGTH;
if (_log.shouldLog(Log.DEBUG)) if (_log.shouldLog(Log.DEBUG))
_log.debug("mode = router, routerHash = " + _routerHash); _log.debug("mode = router, routerHash = " + _routerHash);
break; break;
case FLAG_MODE_TUNNEL: case FLAG_MODE_TUNNEL:
if ( (_routerHash == null) || (_tunnelId == null) ) throw new DataFormatException("Router hash or tunnel ID is not set"); if ( (_routerHash == null) || (_tunnelId == null) ) throw new DataFormatException("Router hash or tunnel ID is not set");
_routerHash.writeBytes(baos); System.arraycopy(_routerHash.getData(), 0, rv, offset, Hash.HASH_LENGTH);
_tunnelId.writeBytes(baos); offset += Hash.HASH_LENGTH;
DataHelper.toLong(rv, offset, 4, _tunnelId.getTunnelId());
offset += 4;
if (_log.shouldLog(Log.DEBUG)) if (_log.shouldLog(Log.DEBUG))
_log.debug("mode = tunnel, tunnelId = " + _tunnelId.getTunnelId() _log.debug("mode = tunnel, tunnelId = " + _tunnelId.getTunnelId()
+ ", routerHash = " + _routerHash); + ", routerHash = " + _routerHash);
@ -261,15 +297,13 @@ public class DeliveryInstructions extends DataStructureImpl {
if (getDelayRequested()) { if (getDelayRequested()) {
if (_log.shouldLog(Log.DEBUG)) if (_log.shouldLog(Log.DEBUG))
_log.debug("delay requested: " + getDelaySeconds()); _log.debug("delay requested: " + getDelaySeconds());
DataHelper.writeLong(baos, 4, getDelaySeconds()); DataHelper.toLong(rv, offset, 4, getDelaySeconds());
offset += 4;
} else { } else {
if (_log.shouldLog(Log.DEBUG)) if (_log.shouldLog(Log.DEBUG))
_log.debug("delay NOT requested"); _log.debug("delay NOT requested");
} }
} catch (IOException ioe) { return rv;
throw new DataFormatException("Unable to write out additional info", ioe);
}
return baos.toByteArray();
} }
public void writeBytes(OutputStream out) throws DataFormatException, IOException { public void writeBytes(OutputStream out) throws DataFormatException, IOException {

24
router/java/src/net/i2p/data/i2np/GarlicClove.java
View File

@ -107,19 +107,29 @@ public class GarlicClove extends DataStructureImpl {
public void writeBytes(OutputStream out) throws DataFormatException, IOException { public void writeBytes(OutputStream out) throws DataFormatException, IOException {
StringBuffer error = new StringBuffer(); StringBuffer error = null;
if (_instructions == null) if (_instructions == null) {
if (error == null) error = new StringBuffer();
error.append("No instructions "); error.append("No instructions ");
if (_msg == null) }
if (_msg == null) {
if (error == null) error = new StringBuffer();
error.append("No message "); error.append("No message ");
if (_cloveId < 0) }
if (_cloveId < 0) {
if (error == null) error = new StringBuffer();
error.append("CloveID < 0 [").append(_cloveId).append("] "); error.append("CloveID < 0 [").append(_cloveId).append("] ");
if (_expiration == null) }
if (_expiration == null) {
if (error == null) error = new StringBuffer();
error.append("Expiration is null "); error.append("Expiration is null ");
if (_certificate == null) }
if (_certificate == null) {
if (error == null) error = new StringBuffer();
error.append("Certificate is null "); error.append("Certificate is null ");
}
if (error.length() > 0) if ( (error != null) && (error.length() > 0) )
throw new DataFormatException(error.toString()); throw new DataFormatException(error.toString());
_instructions.writeBytes(out); _instructions.writeBytes(out);

4
router/java/src/net/i2p/router/RouterVersion.java
View File

@ -15,9 +15,9 @@ import net.i2p.CoreVersion;
* *
*/ */
public class RouterVersion { public class RouterVersion {
public final static String ID = "$Revision: 1.206 $ $Date: 2005/07/20 14:24:47 $"; public final static String ID = "$Revision: 1.207 $ $Date: 2005/07/21 17:37:15 $";
public final static String VERSION = "0.5.0.7"; public final static String VERSION = "0.5.0.7";
public final static long BUILD = 18; public final static long BUILD = 19;
public static void main(String args[]) { public static void main(String args[]) {
System.out.println("I2P Router version: " + VERSION); System.out.println("I2P Router version: " + VERSION);
System.out.println("Router ID: " + RouterVersion.ID); System.out.println("Router ID: " + RouterVersion.ID);

8
router/java/src/net/i2p/router/client/ClientConnectionRunner.java
View File

@ -292,8 +292,8 @@ public class ClientConnectionRunner {
_log.debug("Acking message send [accepted]" + id + " / " + nonce + " for sessionId " _log.debug("Acking message send [accepted]" + id + " / " + nonce + " for sessionId "
+ _sessionId, new Exception("sendAccepted")); + _sessionId, new Exception("sendAccepted"));
MessageStatusMessage status = new MessageStatusMessage(); MessageStatusMessage status = new MessageStatusMessage();
status.setMessageId(id); status.setMessageId(id.getMessageId());
status.setSessionId(_sessionId); status.setSessionId(_sessionId.getSessionId());
status.setSize(0L); status.setSize(0L);
status.setNonce(nonce); status.setNonce(nonce);
status.setStatus(MessageStatusMessage.STATUS_SEND_ACCEPTED); status.setStatus(MessageStatusMessage.STATUS_SEND_ACCEPTED);
@ -491,8 +491,8 @@ public class ClientConnectionRunner {
if (_dead) return; if (_dead) return;
MessageStatusMessage msg = new MessageStatusMessage(); MessageStatusMessage msg = new MessageStatusMessage();
msg.setMessageId(_messageId); msg.setMessageId(_messageId.getMessageId());
msg.setSessionId(_sessionId); msg.setSessionId(_sessionId.getSessionId());
msg.setNonce(2); msg.setNonce(2);
msg.setSize(0); msg.setSize(0);
if (_success) if (_success)

6
router/java/src/net/i2p/router/client/ClientMessageEventListener.java
View File

@ -179,8 +179,8 @@ class ClientMessageEventListener implements I2CPMessageReader.I2CPMessageEventLi
_log.debug("Handling recieve begin: id = " + message.getMessageId()); _log.debug("Handling recieve begin: id = " + message.getMessageId());
MessagePayloadMessage msg = new MessagePayloadMessage(); MessagePayloadMessage msg = new MessagePayloadMessage();
msg.setMessageId(message.getMessageId()); msg.setMessageId(message.getMessageId());
msg.setSessionId(_runner.getSessionId()); msg.setSessionId(_runner.getSessionId().getSessionId());
Payload payload = _runner.getPayload(message.getMessageId()); Payload payload = _runner.getPayload(new MessageId(message.getMessageId()));
if (payload == null) { if (payload == null) {
if (_log.shouldLog(Log.ERROR)) if (_log.shouldLog(Log.ERROR))
_log.error("Payload for message id [" + message.getMessageId() _log.error("Payload for message id [" + message.getMessageId()
@ -202,7 +202,7 @@ class ClientMessageEventListener implements I2CPMessageReader.I2CPMessageEventLi
* *
*/ */
private void handleReceiveEnd(I2CPMessageReader reader, ReceiveMessageEndMessage message) { private void handleReceiveEnd(I2CPMessageReader reader, ReceiveMessageEndMessage message) {
_runner.removePayload(message.getMessageId()); _runner.removePayload(new MessageId(message.getMessageId()));
} }
private void handleDestroySession(I2CPMessageReader reader, DestroySessionMessage message) { private void handleDestroySession(I2CPMessageReader reader, DestroySessionMessage message) {

24
router/java/src/net/i2p/router/client/ClientWriterRunner.java
View File

@ -61,25 +61,27 @@ class ClientWriterRunner implements Runnable {
} }
} }
public void run() { public void run() {
while (!_runner.getIsDead()) { List messages = new ArrayList(64);
List messages = null; List messageTimes = new ArrayList(64);
List messageTimes = null; List switchList = null;
while (!_runner.getIsDead()) {
synchronized (_dataLock) { synchronized (_dataLock) {
if (_messagesToWrite.size() <= 0) if (_messagesToWrite.size() <= 0)
try { _dataLock.wait(); } catch (InterruptedException ie) {} try { _dataLock.wait(); } catch (InterruptedException ie) {}
if (_messagesToWrite.size() > 0) { if (_messagesToWrite.size() > 0) {
messages = new ArrayList(_messagesToWrite.size()); switchList = _messagesToWrite;
messageTimes = new ArrayList(_messagesToWriteTimes.size()); _messagesToWrite = messages;
messages.addAll(_messagesToWrite); messages = switchList;
messageTimes.addAll(_messagesToWriteTimes);
_messagesToWrite.clear(); switchList = _messagesToWriteTimes;
_messagesToWriteTimes.clear(); _messagesToWriteTimes = messageTimes;
messageTimes = switchList;
} }
} }
if (messages != null) { if (messages.size() > 0) {
for (int i = 0; i < messages.size(); i++) { for (int i = 0; i < messages.size(); i++) {
I2CPMessage msg = (I2CPMessage)messages.get(i); I2CPMessage msg = (I2CPMessage)messages.get(i);
Long when = (Long)messageTimes.get(i); Long when = (Long)messageTimes.get(i);
@ -93,6 +95,8 @@ class ClientWriterRunner implements Runnable {
+ msg.getClass().getName()); + msg.getClass().getName());
} }
} }
messages.clear();
messageTimes.clear();
} }
} }
} }

4
router/java/src/net/i2p/router/client/MessageReceivedJob.java
View File

@ -56,8 +56,8 @@ class MessageReceivedJob extends JobImpl {
_log.debug("Sending message available: " + id + " to sessionId " + _runner.getSessionId() _log.debug("Sending message available: " + id + " to sessionId " + _runner.getSessionId()
+ " (with nonce=1)", new Exception("available")); + " (with nonce=1)", new Exception("available"));
MessageStatusMessage msg = new MessageStatusMessage(); MessageStatusMessage msg = new MessageStatusMessage();
msg.setMessageId(id); msg.setMessageId(id.getMessageId());
msg.setSessionId(_runner.getSessionId()); msg.setSessionId(_runner.getSessionId().getSessionId());
msg.setSize(size); msg.setSize(size);
msg.setNonce(1); msg.setNonce(1);
msg.setStatus(MessageStatusMessage.STATUS_AVAILABLE); msg.setStatus(MessageStatusMessage.STATUS_AVAILABLE);

2
router/java/src/net/i2p/router/networkdb/HandleDatabaseLookupMessageJob.java
View File

@ -49,6 +49,7 @@ public class HandleDatabaseLookupMessageJob extends JobImpl {
_log = getContext().logManager().getLog(HandleDatabaseLookupMessageJob.class); _log = getContext().logManager().getLog(HandleDatabaseLookupMessageJob.class);
getContext().statManager().createRateStat("netDb.lookupsHandled", "How many netDb lookups have we handled?", "NetworkDatabase", new long[] { 5*60*1000l, 60*60*1000l, 24*60*60*1000l }); getContext().statManager().createRateStat("netDb.lookupsHandled", "How many netDb lookups have we handled?", "NetworkDatabase", new long[] { 5*60*1000l, 60*60*1000l, 24*60*60*1000l });
getContext().statManager().createRateStat("netDb.lookupsMatched", "How many netDb lookups did we have the data for?", "NetworkDatabase", new long[] { 5*60*1000l, 60*60*1000l, 24*60*60*1000l }); getContext().statManager().createRateStat("netDb.lookupsMatched", "How many netDb lookups did we have the data for?", "NetworkDatabase", new long[] { 5*60*1000l, 60*60*1000l, 24*60*60*1000l });
getContext().statManager().createRateStat("netDb.lookupsMatchedLeaseSet", "How many netDb leaseSet lookups did we have the data for?", "NetworkDatabase", new long[] { 5*60*1000l, 60*60*1000l, 24*60*60*1000l });
getContext().statManager().createRateStat("netDb.lookupsMatchedReceivedPublished", "How many netDb lookups did we have the data for that were published to us?", "NetworkDatabase", new long[] { 5*60*1000l, 60*60*1000l, 24*60*60*1000l }); getContext().statManager().createRateStat("netDb.lookupsMatchedReceivedPublished", "How many netDb lookups did we have the data for that were published to us?", "NetworkDatabase", new long[] { 5*60*1000l, 60*60*1000l, 24*60*60*1000l });
getContext().statManager().createRateStat("netDb.lookupsMatchedLocalClosest", "How many netDb lookups for local data were received where we are the closest peers?", "NetworkDatabase", new long[] { 5*60*1000l, 60*60*1000l, 24*60*60*1000l }); getContext().statManager().createRateStat("netDb.lookupsMatchedLocalClosest", "How many netDb lookups for local data were received where we are the closest peers?", "NetworkDatabase", new long[] { 5*60*1000l, 60*60*1000l, 24*60*60*1000l });
getContext().statManager().createRateStat("netDb.lookupsMatchedLocalNotClosest", "How many netDb lookups for local data were received where we are NOT the closest peers?", "NetworkDatabase", new long[] { 5*60*1000l, 60*60*1000l, 24*60*60*1000l }); getContext().statManager().createRateStat("netDb.lookupsMatchedLocalNotClosest", "How many netDb lookups for local data were received where we are NOT the closest peers?", "NetworkDatabase", new long[] { 5*60*1000l, 60*60*1000l, 24*60*60*1000l });
@ -130,6 +131,7 @@ public class HandleDatabaseLookupMessageJob extends JobImpl {
if (data instanceof LeaseSet) { if (data instanceof LeaseSet) {
msg.setLeaseSet((LeaseSet)data); msg.setLeaseSet((LeaseSet)data);
msg.setValueType(DatabaseStoreMessage.KEY_TYPE_LEASESET); msg.setValueType(DatabaseStoreMessage.KEY_TYPE_LEASESET);
getContext().statManager().addRateData("netDb.lookupsMatchedLeaseSet", 1, 0);
} else if (data instanceof RouterInfo) { } else if (data instanceof RouterInfo) {
msg.setRouterInfo((RouterInfo)data); msg.setRouterInfo((RouterInfo)data);
msg.setValueType(DatabaseStoreMessage.KEY_TYPE_ROUTERINFO); msg.setValueType(DatabaseStoreMessage.KEY_TYPE_ROUTERINFO);

2
router/java/src/net/i2p/router/transport/udp/ACKSender.java
View File

@ -100,6 +100,8 @@ public class ACKSender implements Runnable {
_context.statManager().addRateData("udp.sendACKCount", ackBitfields.size(), 0); _context.statManager().addRateData("udp.sendACKCount", ackBitfields.size(), 0);
_context.statManager().addRateData("udp.sendACKRemaining", remaining, 0); _context.statManager().addRateData("udp.sendACKRemaining", remaining, 0);
now = _context.clock().now(); now = _context.clock().now();
if (lastSend < 0)
lastSend = now - 1;
_context.statManager().addRateData("udp.ackFrequency", now-lastSend, now-wanted); _context.statManager().addRateData("udp.ackFrequency", now-lastSend, now-wanted);
//_context.statManager().getStatLog().addData(peer.getRemoteHostId().toString(), "udp.peer.sendACKCount", ackBitfields.size(), 0); //_context.statManager().getStatLog().addData(peer.getRemoteHostId().toString(), "udp.peer.sendACKCount", ackBitfields.size(), 0);
UDPPacket ack = _builder.buildACK(peer, ackBitfields); UDPPacket ack = _builder.buildACK(peer, ackBitfields);

18
router/java/src/net/i2p/router/transport/udp/PacketHandler.java
View File

@ -41,7 +41,11 @@ public class PacketHandler {
_context.statManager().createRateStat("udp.handleTime", "How long it takes to handle a received packet after its been pulled off the queue", "udp", new long[] { 10*60*1000, 60*60*1000 }); _context.statManager().createRateStat("udp.handleTime", "How long it takes to handle a received packet after its been pulled off the queue", "udp", new long[] { 10*60*1000, 60*60*1000 });
_context.statManager().createRateStat("udp.queueTime", "How long after a packet is received can we begin handling it", "udp", new long[] { 10*60*1000, 60*60*1000 }); _context.statManager().createRateStat("udp.queueTime", "How long after a packet is received can we begin handling it", "udp", new long[] { 10*60*1000, 60*60*1000 });
_context.statManager().createRateStat("udp.receivePacketSkew", "How long ago after the packet was sent did we receive it", "udp", new long[] { 10*60*1000, 60*60*1000 }); _context.statManager().createRateStat("udp.receivePacketSkew", "How long ago after the packet was sent did we receive it", "udp", new long[] { 10*60*1000, 60*60*1000 });
_context.statManager().createRateStat("udp.droppedInvalid", "How old the packet we dropped due to invalidity was", "udp", new long[] { 10*60*1000, 60*60*1000 }); _context.statManager().createRateStat("udp.droppedInvalidUnkown", "How old the packet we dropped due to invalidity (unkown type) was", "udp", new long[] { 10*60*1000, 60*60*1000 });
_context.statManager().createRateStat("udp.droppedInvalidReestablish", "How old the packet we dropped due to invalidity (doesn't use existing key, not an establishment) was", "udp", new long[] { 10*60*1000, 60*60*1000 });
_context.statManager().createRateStat("udp.droppedInvalidEstablish", "How old the packet we dropped due to invalidity (establishment, bad key) was", "udp", new long[] { 10*60*1000, 60*60*1000 });
_context.statManager().createRateStat("udp.droppedInvalidInboundEstablish", "How old the packet we dropped due to invalidity (inbound establishment, bad key) was", "udp", new long[] { 10*60*1000, 60*60*1000 });
_context.statManager().createRateStat("udp.droppedInvalidSkew", "How skewed the packet we dropped due to invalidity (valid except bad skew) was", "udp", new long[] { 10*60*1000, 60*60*1000 });
} }
public void startup() { public void startup() {
@ -158,7 +162,7 @@ public class PacketHandler {
} else { } else {
if (_log.shouldLog(Log.WARN)) if (_log.shouldLog(Log.WARN))
_log.warn("Validation with existing con failed, and validation as reestablish failed too. DROP"); _log.warn("Validation with existing con failed, and validation as reestablish failed too. DROP");
_context.statManager().addRateData("udp.droppedInvalid", packet.getLifetime(), packet.getExpiration()); _context.statManager().addRateData("udp.droppedInvalidReestablish", packet.getLifetime(), packet.getExpiration());
} }
return; return;
} }
@ -177,7 +181,7 @@ public class PacketHandler {
if (!isValid) { if (!isValid) {
if (_log.shouldLog(Log.WARN)) if (_log.shouldLog(Log.WARN))
_log.warn("Invalid introduction packet received: " + packet, new Exception("path")); _log.warn("Invalid introduction packet received: " + packet, new Exception("path"));
_context.statManager().addRateData("udp.droppedInvalid", packet.getLifetime(), packet.getExpiration()); _context.statManager().addRateData("udp.droppedInvalidEstablish", packet.getLifetime(), packet.getExpiration());
return; return;
} else { } else {
if (_log.shouldLog(Log.INFO)) if (_log.shouldLog(Log.INFO))
@ -224,7 +228,7 @@ public class PacketHandler {
// on earlier state packets // on earlier state packets
receivePacket(reader, packet); receivePacket(reader, packet);
} else { } else {
_context.statManager().addRateData("udp.droppedInvalid", packet.getLifetime(), packet.getExpiration()); _context.statManager().addRateData("udp.droppedInvalidInboundEstablish", packet.getLifetime(), packet.getExpiration());
} }
} }
@ -283,12 +287,12 @@ public class PacketHandler {
if (skew > GRACE_PERIOD) { if (skew > GRACE_PERIOD) {
if (_log.shouldLog(Log.WARN)) if (_log.shouldLog(Log.WARN))
_log.warn("Packet too far in the future: " + new Date(sendOn/1000) + ": " + packet); _log.warn("Packet too far in the future: " + new Date(sendOn/1000) + ": " + packet);
_context.statManager().addRateData("udp.droppedInvalid", packet.getLifetime(), packet.getExpiration()); _context.statManager().addRateData("udp.droppedInvalidSkew", skew, packet.getExpiration());
return; return;
} else if (skew < 0 - GRACE_PERIOD) { } else if (skew < 0 - GRACE_PERIOD) {
if (_log.shouldLog(Log.WARN)) if (_log.shouldLog(Log.WARN))
_log.warn("Packet too far in the past: " + new Date(sendOn/1000) + ": " + packet); _log.warn("Packet too far in the past: " + new Date(sendOn/1000) + ": " + packet);
_context.statManager().addRateData("udp.droppedInvalid", packet.getLifetime(), packet.getExpiration()); _context.statManager().addRateData("udp.droppedInvalidSkew", 0-skew, packet.getExpiration());
return; return;
} }
@ -324,7 +328,7 @@ public class PacketHandler {
default: default:
if (_log.shouldLog(Log.WARN)) if (_log.shouldLog(Log.WARN))
_log.warn("Unknown payload type: " + reader.readPayloadType()); _log.warn("Unknown payload type: " + reader.readPayloadType());
_context.statManager().addRateData("udp.droppedInvalid", packet.getLifetime(), packet.getExpiration()); _context.statManager().addRateData("udp.droppedInvalidUnknown", packet.getLifetime(), packet.getExpiration());
return; return;
} }
} }

6
router/java/src/net/i2p/router/transport/udp/UDPSender.java
View File

@ -168,9 +168,11 @@ public class UDPSender {
} }
long sendTime = _context.clock().now() - before; long sendTime = _context.clock().now() - before;
_context.statManager().addRateData("udp.socketSendTime", sendTime, packet.getLifetime()); _context.statManager().addRateData("udp.socketSendTime", sendTime, packet.getLifetime());
_context.statManager().addRateData("udp.sendBWThrottleTime", afterBW - acquireTime, acquireTime - packet.getBegin()); long throttleTime = afterBW - acquireTime;
if (throttleTime > 10)
_context.statManager().addRateData("udp.sendBWThrottleTime", throttleTime, acquireTime - packet.getBegin());
if (packet.getMarkedType() == 1) if (packet.getMarkedType() == 1)
_context.statManager().addRateData("udp.sendACKTime", afterBW - acquireTime, packet.getLifetime()); _context.statManager().addRateData("udp.sendACKTime", throttleTime, packet.getLifetime());
_context.statManager().addRateData("udp.pushTime", packet.getLifetime(), packet.getLifetime()); _context.statManager().addRateData("udp.pushTime", packet.getLifetime(), packet.getLifetime());
_context.statManager().addRateData("udp.sendPacketSize", size, packet.getLifetime()); _context.statManager().addRateData("udp.sendPacketSize", size, packet.getLifetime());
} catch (IOException ioe) { } catch (IOException ioe) {

9
router/java/src/net/i2p/router/transport/udp/UDPTransport.java
View File

@ -128,6 +128,7 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
_expireEvent = new ExpirePeerEvent(); _expireEvent = new ExpirePeerEvent();
_context.statManager().createRateStat("udp.droppedPeer", "How long ago did we receive from a dropped peer (duration == session lifetime", "udp", new long[] { 60*60*1000, 24*60*60*1000 }); _context.statManager().createRateStat("udp.droppedPeer", "How long ago did we receive from a dropped peer (duration == session lifetime", "udp", new long[] { 60*60*1000, 24*60*60*1000 });
_context.statManager().createRateStat("udp.droppedPeerInactive", "How long ago did we receive from a dropped peer (duration == session lifetime)", "udp", new long[] { 60*60*1000, 24*60*60*1000 });
} }
public void startup() { public void startup() {
@ -315,6 +316,7 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
boolean addRemotePeerState(PeerState peer) { boolean addRemotePeerState(PeerState peer) {
if (_log.shouldLog(Log.INFO)) if (_log.shouldLog(Log.INFO))
_log.info("Add remote peer state: " + peer); _log.info("Add remote peer state: " + peer);
long oldEstablishedOn = -1;
PeerState oldPeer = null; PeerState oldPeer = null;
if (peer.getRemotePeer() != null) { if (peer.getRemotePeer() != null) {
synchronized (_peersByIdent) { synchronized (_peersByIdent) {
@ -323,6 +325,7 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
// should we transfer the oldPeer's RTT/RTO/etc? nah // should we transfer the oldPeer's RTT/RTO/etc? nah
// or perhaps reject the new session? nah, // or perhaps reject the new session? nah,
// using the new one allow easier reconnect // using the new one allow easier reconnect
oldEstablishedOn = oldPeer.getKeyEstablishedTime();
} }
} }
} }
@ -339,6 +342,7 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
if ( (oldPeer != null) && (oldPeer != peer) ) { if ( (oldPeer != null) && (oldPeer != peer) ) {
//_peersByRemoteHost.put(remoteString, oldPeer); //_peersByRemoteHost.put(remoteString, oldPeer);
//return false; //return false;
oldEstablishedOn = oldPeer.getKeyEstablishedTime();
} }
} }
@ -353,6 +357,8 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
_expireEvent.add(peer); _expireEvent.add(peer);
if (oldEstablishedOn > 0)
_context.statManager().addRateData("udp.alreadyConnected", oldEstablishedOn, 0);
return true; return true;
} }
@ -367,6 +373,9 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
long now = _context.clock().now(); long now = _context.clock().now();
_context.statManager().addRateData("udp.droppedPeer", now - peer.getLastReceiveTime(), now - peer.getKeyEstablishedTime()); _context.statManager().addRateData("udp.droppedPeer", now - peer.getLastReceiveTime(), now - peer.getKeyEstablishedTime());
_context.shitlist().shitlistRouter(peer.getRemotePeer(), "dropped after too many retries"); _context.shitlist().shitlistRouter(peer.getRemotePeer(), "dropped after too many retries");
} else {
long now = _context.clock().now();
_context.statManager().addRateData("udp.droppedPeerInactive", now - peer.getLastReceiveTime(), now - peer.getKeyEstablishedTime());
} }
synchronized (_peersByIdent) { synchronized (_peersByIdent) {
_peersByIdent.remove(peer.getRemotePeer()); _peersByIdent.remove(peer.getRemotePeer());