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

propagate from branch 'i2p.i2p.zzz.test2' (head d80f8632ef0a2aa350eee9a38ba6a4c56471f5bc)

Browse Source 
to branch 'i2p.i2p' (head e8b712c826e44211217f06eb3a4da2703323a42f)
This commit is contained in:
zzz
2013-07-20 11:53:41 +00:00
parent 3b46acc285 554a3a6b0e
commit 90cc71d14d
37 changed files with 14820 additions and 101278 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
apps/streaming/java/src/net/i2p/client/streaming/ConnectionOptions.java
View File

@@ -27,7 +27,7 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
private int _profile;
private int _rtt;
private int _rttDev;
private int _rto;
private int _rto = INITIAL_RTO;
private int _resendDelay;
private int _sendAckDelay;
private int _maxMessageSize;
@@ -51,7 +51,17 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
private int _maxTotalConnsPerDay;
private int _maxConns;
private boolean _disableRejectLog;
/** state of a connection */
private enum AckInit {
INIT, // just created
FIRST, // first received ack
STEADY
}
/** LOCKING: this */
private AckInit _initState = AckInit.INIT;
// NOTE - almost all the options are below, but see
// I2PSocketOptions in ministreaming for a few more
@@ -65,11 +75,21 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
/** on inactivity timeout, send a payload message */
public static final int INACTIVITY_ACTION_SEND = 2;
/*
* These values are specified in RFC 6298
* Do not change unless you know what you're doing
*/
private static final double TCP_ALPHA = 1.0/8;
private static final double TCP_BETA = 1.0/4;
private static final double TCP_KAPPA = 4;
private static final String PROP_INITIAL_RTO = "i2p.streaming.initialRTO";
private static final int INITIAL_RTO = 12000;
public static final String PROP_CONNECT_DELAY = "i2p.streaming.connectDelay";
public static final String PROP_PROFILE = "i2p.streaming.profile";
public static final String PROP_MAX_MESSAGE_SIZE = "i2p.streaming.maxMessageSize";
public static final String PROP_MAX_RESENDS = "i2p.streaming.maxResends";
public static final String PROP_INITIAL_RTT = "i2p.streaming.initialRTT";
public static final String PROP_INITIAL_RESEND_DELAY = "i2p.streaming.initialResendDelay";
public static final String PROP_INITIAL_ACK_DELAY = "i2p.streaming.initialAckDelay";
public static final String PROP_INITIAL_WINDOW_SIZE = "i2p.streaming.initialWindowSize";
@@ -295,6 +315,7 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
setMaxWindowSize(opts.getMaxWindowSize());
setConnectDelay(opts.getConnectDelay());
setProfile(opts.getProfile());
setRTTDev(opts.getRTTDev());
setRTT(opts.getRTT());
setRequireFullySigned(opts.getRequireFullySigned());
setWindowSize(opts.getWindowSize());
@@ -332,7 +353,6 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
setConnectDelay(getInt(opts, PROP_CONNECT_DELAY, -1));
setProfile(getInt(opts, PROP_PROFILE, PROFILE_BULK));
setMaxMessageSize(getInt(opts, PROP_MAX_MESSAGE_SIZE, DEFAULT_MAX_MESSAGE_SIZE));
setRTT(getInt(opts, PROP_INITIAL_RTT, DEFAULT_INITIAL_RTT));
setReceiveWindow(getInt(opts, PROP_INITIAL_RECEIVE_WINDOW, 1));
setResendDelay(getInt(opts, PROP_INITIAL_RESEND_DELAY, 1000));
setSendAckDelay(getInt(opts, PROP_INITIAL_ACK_DELAY, DEFAULT_INITIAL_ACK_DELAY));
@@ -360,6 +380,8 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
_maxTotalConnsPerHour = getInt(opts, PROP_MAX_TOTAL_CONNS_HOUR, 0);
_maxTotalConnsPerDay = getInt(opts, PROP_MAX_TOTAL_CONNS_DAY, 0);
_maxConns = getInt(opts, PROP_MAX_STREAMS, 0);
_rto = getInt(opts, PROP_INITIAL_RTO, INITIAL_RTO);
}
/**
@@ -377,8 +399,6 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
setProfile(getInt(opts, PROP_PROFILE, PROFILE_BULK));
if (opts.containsKey(PROP_MAX_MESSAGE_SIZE))
setMaxMessageSize(getInt(opts, PROP_MAX_MESSAGE_SIZE, Packet.MAX_PAYLOAD_SIZE));
if (opts.containsKey(PROP_INITIAL_RTT))
setRTT(getInt(opts, PROP_INITIAL_RTT, DEFAULT_INITIAL_RTT));
if (opts.containsKey(PROP_INITIAL_RECEIVE_WINDOW))
setReceiveWindow(getInt(opts, PROP_INITIAL_RECEIVE_WINDOW, 1));
if (opts.containsKey(PROP_INITIAL_RESEND_DELAY))
@@ -427,6 +447,8 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
_maxTotalConnsPerDay = getInt(opts, PROP_MAX_TOTAL_CONNS_DAY, 0);
if (opts.containsKey(PROP_MAX_STREAMS))
_maxConns = getInt(opts, PROP_MAX_STREAMS, 0);
_rto = getInt(opts, PROP_INITIAL_RTO, INITIAL_RTO);
}
/**
@@ -515,12 +537,8 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
* What to set the round trip time estimate to (in milliseconds)
* @return round trip time estimate in ms
*/
public int getRTT() { return _rtt; }
public synchronized int getRTT() { return _rtt; }
public void setRTT(int ms) {
if (_rto == 0) {
_rttDev = ms / 2;
_rto = ms + ms / 2;
}
synchronized (_trend) {
_trend[0] = _trend[1];
_trend[1] = _trend[2];
@@ -532,15 +550,50 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
_trend[2] = 0;
}
_rtt = ms;
if (_rtt > 60*1000)
_rtt = 60*1000;
synchronized(this) {
_rtt = ms;
if (_rtt > 60*1000)
_rtt = 60*1000;
}
}
public int getRTO() { return _rto; }
public synchronized int getRTO() { return _rto; }
/** for debugging @since 0.7.13 */
int getRTTDev() { return _rttDev; }
/** used in TCB @since 0.9.8 */
synchronized int getRTTDev() { return _rttDev; }
private synchronized void setRTTDev(int rttDev) { _rttDev = rttDev; }
/**
* Loads options from TCB cache.
*/
synchronized void loadFromCache(int rtt, int rttDev, int wdw) {
_initState = AckInit.STEADY;
setRTT(rtt);
setRTTDev(rttDev);
setWindowSize(wdw);
computeRTO();
}
/**
* computes RTO based on formula in RFC
*/
private synchronized void computeRTO() {
switch(_initState) {
case INIT :
throw new IllegalStateException();
case FIRST :
_rto = _rtt + _rtt / 2;
break;
case STEADY :
_rto = _rtt + (int) (_rttDev * TCP_KAPPA);
break;
}
if (_rto < Connection.MIN_RESEND_DELAY)
_rto = (int)Connection.MIN_RESEND_DELAY;
else if (_rto > Connection.MAX_RESEND_DELAY)
_rto = (int)Connection.MAX_RESEND_DELAY;
}
/**
* If we have 3 consecutive rtt increases, we are trending upwards (1), or if we have
@@ -558,22 +611,22 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
}
}
/** rtt = rtt*RTT_DAMPENING + (1-RTT_DAMPENING)*currentPacketRTT */
/** This is the value specified in RFC 2988, let's try it */
private static final double RTT_DAMPENING = 0.875;
public void updateRTT(int measuredValue) {
// the rttDev calculation matches that recommended in RFC 2988 (beta = 1/4)
_rttDev = _rttDev + (int)(0.25d*(Math.abs(measuredValue-_rtt)-_rttDev));
int smoothed = (int)(RTT_DAMPENING*_rtt + (1-RTT_DAMPENING)*measuredValue);
// K = 4
_rto = smoothed + (_rttDev<<2);
if (_rto < Connection.MIN_RESEND_DELAY)
_rto = (int)Connection.MIN_RESEND_DELAY;
else if (_rto > Connection.MAX_RESEND_DELAY)
_rto = (int)Connection.MAX_RESEND_DELAY;
setRTT(smoothed);
public synchronized void updateRTT(int measuredValue) {
switch(_initState) {
case INIT:
_initState = AckInit.FIRST;
setRTT(measuredValue); // no smoothing first sample
_rttDev = _rtt / 2;
break;
case FIRST:
_initState = AckInit.STEADY; // fall through
case STEADY:
// calculation matches that recommended in RFC 6298
_rttDev = (int) ((1-TCP_BETA) *_rttDev + TCP_BETA * Math.abs(measuredValue-_rtt));
int smoothed = (int)((1-TCP_ALPHA)*_rtt + TCP_ALPHA*measuredValue);
setRTT(smoothed);
}
computeRTO();
}
/** How long after sending a packet will we wait before resending?

66
apps/streaming/java/src/net/i2p/client/streaming/TCBShare.java
View File

@@ -2,6 +2,7 @@ package net.i2p.client.streaming;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import net.i2p.I2PAppContext;
@@ -9,6 +10,8 @@ import net.i2p.data.Destination;
import net.i2p.util.Log;
import net.i2p.util.SimpleTimer2;
import static net.i2p.client.streaming.I2PSocketOptionsImpl.getDouble;
/**
* Share important TCP Control Block parameters across Connections
* to the same remote peer.
@@ -25,20 +28,43 @@ class TCBShare {
private final Log _log;
private final Map<Destination, Entry> _cache;
private final CleanEvent _cleaner;
private final double _rttDampening, _wdwDampening, _rttDevDampening;
private static final long EXPIRE_TIME = 30*60*1000;
private static final long CLEAN_TIME = 10*60*1000;
///// constants defined in rfc 2140
///// do not change unless you know what you're doing
private static final double RTT_DAMPENING = 0.75;
private static final double RTTDEV_DAMPENING = 0.75;
private static final double WDW_DAMPENING = 0.75;
private static final String RTT_DAMP_PROP="i2p.streaming.tcbcache.rttDampening";
private static final String WDW_DAMP_PROP="i2p.streaming.tcbcache.wdwDampening";
private static final String RTTDEV_DAMP_PROP="i2p.streaming.tcbcache.rttdevDampening";
/////
private static final int MAX_RTT = ((int) Connection.MAX_RESEND_DELAY) / 2;
private static final int MAX_RTT_DEV = (int) (MAX_RTT * 1.5);
private static final int MAX_WINDOW_SIZE = ConnectionPacketHandler.MAX_SLOW_START_WINDOW;
public TCBShare(I2PAppContext ctx, SimpleTimer2 timer) {
_context = ctx;
_log = ctx.logManager().getLog(TCBShare.class);
final Properties props = ctx.getProperties();
_rttDampening = getDouble(props, RTT_DAMP_PROP, RTT_DAMPENING);
_wdwDampening = getDouble(props, WDW_DAMP_PROP, WDW_DAMPENING);
_rttDevDampening = getDouble(props, RTTDEV_DAMP_PROP, RTTDEV_DAMPENING);
_cache = new ConcurrentHashMap<Destination,Entry>(4);
_cleaner = new CleanEvent(timer);
_cleaner.schedule(CLEAN_TIME);
if (_log.shouldLog(Log.DEBUG)) {
String log = "Creating TCBCache with rttDamp=%s, rttDevDamp=%s, wdwDamp=%s, "+
"expire=%d, clean=%d";
log = String.format(log,_rttDampening,_rttDevDampening,_wdwDampening,
EXPIRE_TIME,CLEAN_TIME);
_log.debug(log);
}
}
/**
@@ -60,14 +86,22 @@ class TCBShare {
Entry e = _cache.get(dest);
if (e == null || e.isExpired())
return;
if (_log.shouldLog(Log.DEBUG))
final int rtt, rttDev, wdw;
synchronized(e) {
rtt = e.getRTT();
rttDev = e.getRTTDev();
wdw = e.getWindowSize();
}
if (_log.shouldLog(Log.DEBUG)) {
_log.debug("From cache: " +
con.getSession().getMyDestination().calculateHash().toBase64().substring(0, 4) +
'-' +
dest.calculateHash().toBase64().substring(0, 4) +
" RTT: " + e.getRTT() + " wdw: " + e.getWindowSize());
opts.setRTT(e.getRTT());
opts.setWindowSize(e.getWindowSize());
" RTT: " + rtt +
" RTTDev: "+ rttDev +
" wdw: " + wdw );
}
opts.loadFromCache(rtt,rttDev,wdw);
}
/** store to cache */
@@ -82,47 +116,61 @@ class TCBShare {
return;
int old = -1;
int oldw = -1;
int oldDev = -1;
Entry e = _cache.get(dest);
if (e == null || e.isExpired()) {
e = new Entry(opts.getRTT(), opts.getWindowSize());
e = new Entry(opts.getRTT(), opts.getWindowSize(), opts.getRTTDev());
_cache.put(dest, e);
} else {
synchronized(e) {
old = e.getRTT();
oldw = e.getWindowSize();
oldDev = e.getRTTDev();
e.setRTT(opts.getRTT());
e.setWindowSize(opts.getWindowSize());
e.setRTTDev(opts.getRTTDev());
}
}
if (_log.shouldLog(Log.DEBUG))
if (_log.shouldLog(Log.DEBUG)) {
_log.debug("To cache: " +
con.getSession().getMyDestination().calculateHash().toBase64().substring(0, 4) +
'-' +
dest.calculateHash().toBase64().substring(0, 4) +
" old: " + old + " con: " + opts.getRTT() + " new: " + e.getRTT() +
" oldDev: " + oldDev + " conDev: " + opts.getRTTDev() + " newDev: " + e.getRTTDev() +
" oldw: " + oldw + " conw: " + opts.getWindowSize() + " neww: " + e.getWindowSize());
}
}
private class Entry {
int _rtt;
int _wdw;
int _rttDev;
long _updated;
public Entry(int ms, int wdw) {
public Entry(int ms, int wdw, int rttDev) {
_rtt = ms;
_wdw = wdw;
_rttDev = rttDev;
_updated = _context.clock().now();
}
public synchronized int getRTT() { return _rtt; }
public synchronized void setRTT(int ms) {
_rtt = (int)(RTT_DAMPENING*_rtt + (1-RTT_DAMPENING)*ms);
_rtt = (int)(_rttDampening*_rtt + (1-_rttDampening)*ms);
if (_rtt > MAX_RTT)
_rtt = MAX_RTT;
_updated = _context.clock().now();
}
public synchronized int getRTTDev() { return _rttDev; }
public synchronized void setRTTDev(int count) {
_rttDev = (int)(_rttDevDampening*_rttDev + (1-_rttDevDampening)*count);
if (_rttDev > MAX_RTT_DEV)
_rttDev = MAX_RTT_DEV;
_updated = _context.clock().now();
}
public synchronized int getWindowSize() { return _wdw; }
public synchronized void setWindowSize(int wdw) {
_wdw = (int)(0.5 + WDW_DAMPENING*_wdw + (1-WDW_DAMPENING)*wdw);
_wdw = (int)(0.5 + _wdwDampening*_wdw + (1-_wdwDampening)*wdw);
if (_wdw > MAX_WINDOW_SIZE)
_wdw = MAX_WINDOW_SIZE;
_updated = _context.clock().now();