zzz/i2p.i2p
1
0
Fork 0
forked from I2P_Developers/i2p.i2p
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Streaming read timeout fixes:

Browse Source 
i2ptunnel:
- Better timeout handling when reading headers in HTTP server (improved fix for ticket #723)
  Enforce total header timeout with new readLine()
- Prep for returning specific HTTP errors to client on request timeout and header errors,
  instead of just closing socket... further work to be in i2p.i2p.zzz.test2 branch
Streaming:
- Fix read timeout on input stream - was waiting too long, often twice as long as timeout, or more
  Enforce total timeout even when notify()ed
- Fix read() returning 0 on read timeout instead of -1 (possible fix for ticket #335)
  This prevents passing partial headers to server on timeout
- Fix javadocs for read timeout to match current behavior
- Fix StandardSocket SoTimeout to account for differences with I2PSocket readTimeout
- log tweaks
This commit is contained in:
zzz
2015-04-04 17:00:57 +00:00
parent 351a1a8d27
commit 706ee243a5
8 changed files with 181 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -51,6 +51,8 @@ class I2PSocketFull implements I2PSocket {
}
Connection c = _connection;
if (c == null) return;
if (log.shouldLog(Log.INFO))
log.info("close() called, connected? " + c.getIsConnected() + " : " + c);
if (c.getIsConnected()) {
MessageInputStream in = c.getInputStream();
in.close();
@@ -136,6 +138,8 @@ class I2PSocketFull implements I2PSocket {
Connection c = _connection;
if (c == null) return;
if (ms > Integer.MAX_VALUE)
ms = Integer.MAX_VALUE;
c.getInputStream().setReadTimeout((int)ms);
c.getOptions().setReadTimeout(ms);
}

8
apps/streaming/java/src/net/i2p/client/streaming/impl/I2PSocketOptionsImpl.java
View File

@@ -144,9 +144,11 @@ class I2PSocketOptionsImpl implements I2PSocketOptions {
/**
* What is the longest we'll block on the input stream while waiting
* for more data. If this value is exceeded, the read() throws
* InterruptedIOException
* InterruptedIOException - FIXME doesn't really, returns -1 or 0 instead.
*
* WARNING: Default -1 (unlimited), which is probably not what you want.
*
* @return timeout in ms, 0 for nonblocking, -1 for forever
*/
public long getReadTimeout() {
return _readTimeout;
@@ -155,9 +157,11 @@ class I2PSocketOptionsImpl implements I2PSocketOptions {
/**
* What is the longest we'll block on the input stream while waiting
* for more data. If this value is exceeded, the read() throws
* InterruptedIOException
* InterruptedIOException - FIXME doesn't really, returns -1 or 0 instead.
*
* WARNING: Default -1 (unlimited), which is probably not what you want.
*
* @param ms timeout in ms, 0 for nonblocking, -1 for forever
*/
public void setReadTimeout(long ms) {
_readTimeout = ms;

75
apps/streaming/java/src/net/i2p/client/streaming/impl/MessageInputStream.java
View File

@@ -170,10 +170,15 @@ class MessageInputStream extends InputStream {
/**
* how long a read() call should block (if less than 0, block indefinitely,
* but if it is 0, do not block at all)
* @return how long read calls should block, 0 or less indefinitely block
* @return how long read calls should block, 0 for nonblocking, negative to indefinitely block
*/
public int getReadTimeout() { return _readTimeout; }
/**
* how long a read() call should block (if less than 0, block indefinitely,
* but if it is 0, do not block at all)
* @param timeout how long read calls should block, 0 for nonblocking, negative to indefinitely block
*/
public void setReadTimeout(int timeout) {
if (_log.shouldLog(Log.DEBUG))
_log.debug("Changing read timeout from " + _readTimeout + " to " + timeout);
@@ -230,7 +235,8 @@ class MessageInputStream extends InputStream {
*/
public boolean messageReceived(long messageId, ByteArray payload) {
if (_log.shouldLog(Log.DEBUG))
_log.debug("received " + messageId + " with " + (payload != null ? payload.getValid()+"" : "no payload"));
_log.debug("received msg ID " + messageId + " with " +
(payload != null ? payload.getValid() + " bytes" : "no payload"));
synchronized (_dataLock) {
if (messageId <= _highestReadyBlockId) {
if (_log.shouldLog(Log.INFO))
@@ -261,6 +267,10 @@ class MessageInputStream extends InputStream {
cur++;
_highestReadyBlockId++;
}
// FIXME Javadocs for setReadTimeout() say we will throw
// an InterruptedIOException.
// Java throws a SocketTimeoutException.
// We do neither.
} else {
if (_log.shouldLog(Log.INFO))
_log.info("Message is out of order: " + messageId);
@@ -275,23 +285,41 @@ class MessageInputStream extends InputStream {
return true;
}
/**
* On a read timeout, this returns -1
* (doesn't throw SocketTimeoutException like Socket)
* (doesn't throw InterruptedIOException like our javadocs say)
*/
public int read() throws IOException {
int read = read(_oneByte, 0, 1);
if (read < 0)
if (read <= 0)
return -1;
return _oneByte[0] & 0xff;
}
/**
* On a read timeout, this returns 0
* (doesn't throw SocketTimeoutException like Socket)
* (doesn't throw InterruptedIOException like our javadocs say)
*/
@Override
public int read(byte target[]) throws IOException {
return read(target, 0, target.length);
}
/**
* On a read timeout, this returns 0
* (doesn't throw SocketTimeoutException like Socket)
* (doesn't throw InterruptedIOException like our javadocs say)
*/
@Override
public int read(byte target[], int offset, int length) throws IOException {
long expiration = -1;
if (_readTimeout > 0)
expiration = _readTimeout + System.currentTimeMillis();
int readTimeout = _readTimeout;
long expiration;
if (readTimeout > 0)
expiration = readTimeout + System.currentTimeMillis();
else
expiration = -1;
synchronized (_dataLock) {
if (_locallyClosed) throw new IOException("Already locally closed");
throwAnyError();
@@ -310,10 +338,10 @@ class MessageInputStream extends InputStream {
+ "] got EOF after " + _readTotal + " " + toString());
return -1;
} else {
if (_readTimeout < 0) {
if (readTimeout < 0) {
if (_log.shouldLog(Log.DEBUG))
_log.debug("read(...," + offset+", " + length+ ")[" + i
+ ") with no timeout: " + toString());
+ "] with no timeout: " + toString());
try {
_dataLock.wait();
} catch (InterruptedException ie) {
@@ -323,14 +351,14 @@ class MessageInputStream extends InputStream {
}
if (_log.shouldLog(Log.DEBUG))
_log.debug("read(...," + offset+", " + length+ ")[" + i
+ ") with no timeout complete: " + toString());
+ "] with no timeout complete: " + toString());
throwAnyError();
} else if (_readTimeout > 0) {
} else if (readTimeout > 0) {
if (_log.shouldLog(Log.DEBUG))
_log.debug("read(...," + offset+", " + length+ ")[" + i
+ ") with timeout: " + _readTimeout + ": " + toString());
+ "] with timeout: " + readTimeout + ": " + toString());
try {
_dataLock.wait(_readTimeout);
_dataLock.wait(readTimeout);
} catch (InterruptedException ie) {
IOException ioe2 = new InterruptedIOException("Interrupted read");
ioe2.initCause(ie);
@@ -338,21 +366,30 @@ class MessageInputStream extends InputStream {
}
if (_log.shouldLog(Log.DEBUG))
_log.debug("read(...," + offset+", " + length+ ")[" + i
+ ") with timeout complete: " + _readTimeout + ": " + toString());
+ "] with timeout complete: " + readTimeout + ": " + toString());
throwAnyError();
} else { // readTimeout == 0
// noop, don't block
if (_log.shouldLog(Log.DEBUG))
_log.debug("read(...," + offset+", " + length+ ")[" + i
+ ") with nonblocking setup: " + toString());
+ "] with nonblocking setup: " + toString());
return i;
}
if (_readyDataBlocks.isEmpty()) {
if ( (_readTimeout > 0) && (expiration < System.currentTimeMillis()) ) {
if (_log.shouldLog(Log.INFO))
_log.info("read(...," + offset+", " + length+ ")[" + i
+ ") expired: " + toString());
return i;
if (readTimeout > 0) {
long remaining = expiration - System.currentTimeMillis();
if (remaining <= 0) {
// FIXME Javadocs for setReadTimeout() say we will throw
// an InterruptedIOException.
// Java throws a SocketTimeoutException.
// We do neither.
if (_log.shouldLog(Log.INFO))
_log.info("read(...," + offset+", " + length+ ")[" + i
+ "] expired: " + toString());
return i;
} else {
readTimeout = (int) remaining;
}
}
}
}

13
apps/streaming/java/src/net/i2p/client/streaming/impl/StandardSocket.java
View File

@@ -193,7 +193,15 @@ class StandardSocket extends Socket {
I2PSocketOptions opts = _socket.getOptions();
if (opts == null)
return 0;
return (int) opts.getReadTimeout();
long rv = opts.getReadTimeout();
// Java Socket: 0 is forever, and we don't exactly have nonblocking
if (rv > Integer.MAX_VALUE)
rv = Integer.MAX_VALUE;
else if (rv < 0)
rv = 0;
else if (rv == 0)
rv = 1;
return (int) rv;
}
/**
@@ -309,6 +317,9 @@ class StandardSocket extends Socket {
I2PSocketOptions opts = _socket.getOptions();
if (opts == null)
throw new SocketException("No options");
// Java Socket: 0 is forever
if (timeout == 0)
timeout = -1;
opts.setReadTimeout(timeout);
}