I2P_Developers/i2p.itoopie
3
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

propagate from branch 'i2p.i2p.zzz.test3' (head 75804c9355329ab942a5e2e8ecc91e2ec1eb9d33)

Browse Source 
to branch 'i2p.i2p' (head 33e55422045b9b3c8ae410162f5cca3eb9baae49)
This commit is contained in:
zzz
2009-08-27 19:55:56 +00:00
parent ed0575e937 2e2bff3f0a
commit ebf5eb6e20
45 changed files with 978 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
core/java/src/gnu/crypto/prng/AsyncFortunaStandalone.java
View File

@ -12,7 +12,11 @@ import net.i2p.util.Log;
* has been eaten)
*/
public class AsyncFortunaStandalone extends FortunaStandalone implements Runnable {
private static final int DEFAULT_BUFFERS = 16;
/**
* This is set to 2 to minimize memory usage for standalone apps.
* The router must override this via the prng.buffers property in the router context.
*/
private static final int DEFAULT_BUFFERS = 2;
private static final int BUFSIZE = 256*1024;
private int _bufferCount;
private final byte asyncBuffers[][];
@ -28,7 +32,7 @@ public class AsyncFortunaStandalone extends FortunaStandalone implements Runnabl
public AsyncFortunaStandalone(I2PAppContext context) {
super();
_bufferCount = context.getProperty("router.prng.buffers", DEFAULT_BUFFERS);
_bufferCount = Math.max(context.getProperty("prng.buffers", DEFAULT_BUFFERS), 2);
asyncBuffers = new byte[_bufferCount][BUFSIZE];
status = new int[_bufferCount];
for (int i = 0; i < _bufferCount; i++)

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

@ -283,7 +283,10 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
if (_log.shouldLog(Log.DEBUG)) _log.debug(getPrefix() + "Before getDate");
sendMessage(new GetDateMessage());
if (_log.shouldLog(Log.DEBUG)) _log.debug(getPrefix() + "After getDate / begin waiting for a response");
int waitcount = 0;
while (!_dateReceived) {
if (waitcount++ > 30)
throw new IOException("no date handshake");
try {
synchronized (_dateReceivedLock) {
_dateReceivedLock.wait(1000);
@ -298,7 +301,10 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
if (_log.shouldLog(Log.DEBUG)) _log.debug(getPrefix() + "After producer.connect()");
// wait until we have created a lease set
waitcount = 0;
while (_leaseSet == null) {
if (waitcount++ > 5*60)
throw new IOException("no leaseset");
synchronized (_leaseSetWait) {
try {
_leaseSetWait.wait(1000);

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

@ -91,6 +91,8 @@ class I2PSessionImpl2 extends I2PSessionImpl {
* Perhaps the http server (which does its own compression)
* and P2P apps (with generally uncompressible data) should
* set to false.
*
* Todo: don't compress if destination is local?
*/
private static final int DONT_COMPRESS_SIZE = 66;
protected boolean shouldCompress(int size) {

12
core/java/src/net/i2p/data/DataHelper.java
View File

@ -798,12 +798,20 @@ public class DataHelper {
*
*/
public static int hashCode(byte b[]) {
// Java 5 now has its own method, and the old way
// is horrible for arrays smaller than 32.
// otoh, for sizes >> 32, java's method may be too slow
int rv = 0;
if (b != null) {
for (int i = 0; i < b.length && i < 32; i++)
rv += (b[i] << i);
if (b.length <= 32) {
rv = Arrays.hashCode(b);
} else {
for (int i = 0; i < b.length && i < 32; i++)
rv ^= (b[i] << i); // xor better than + in tests
}
}
return rv;
}
/**

6
core/java/src/net/i2p/data/Destination.java
View File

@ -137,10 +137,12 @@ public class Destination extends DataStructureImpl {
&& DataHelper.eq(getPublicKey(), dst.getPublicKey());
}
/** the public key has enough randomness in it to use it by itself for speed */
@Override
public int hashCode() {
return DataHelper.hashCode(getCertificate()) + DataHelper.hashCode(getSigningPublicKey())
+ DataHelper.hashCode(getPublicKey());
if (_publicKey == null)
return 0;
return _publicKey.hashCode();
}
@Override

10
core/java/src/net/i2p/data/Hash.java
View File

@ -147,9 +147,15 @@ public class Hash extends DataStructureImpl {
return DataHelper.eq(_data, ((Hash) obj)._data);
}
/** a Hash is a hash, so just use the first 4 bytes for speed */
@Override
public int hashCode() {
return DataHelper.hashCode(_data);
int rv = 0;
if (_data != null) {
for (int i = 0; i < 4; i++)
rv ^= (_data[i] << (i*8));
}
return rv;
}
@Override
@ -267,4 +273,4 @@ public class Hash extends DataStructureImpl {
}
_log.debug("Fill check test passed");
}
}
}

8
core/java/src/net/i2p/data/LeaseSet.java
View File

@ -345,12 +345,12 @@ public class LeaseSet extends DataStructureImpl {
}
/** the destination has enough randomness in it to use it by itself for speed */
@Override
public int hashCode() {
return DataHelper.hashCode(getEncryptionKey()) +
//(int)_version +
DataHelper.hashCode(_leases) + DataHelper.hashCode(getSignature())
+ DataHelper.hashCode(getSigningKey()) + DataHelper.hashCode(getDestination());
if (_destination == null)
return 0;
return _destination.hashCode();
}
@Override

10
core/java/src/net/i2p/data/PrivateKey.java
View File

@ -70,9 +70,15 @@ public class PrivateKey extends DataStructureImpl {
return DataHelper.eq(_data, ((PrivateKey) obj)._data);
}
/** the key has enough randomness in it, use the first 4 bytes for speed */
@Override
public int hashCode() {
return DataHelper.hashCode(_data);
int rv = 0;
if (_data != null) {
for (int i = 0; i < 4; i++)
rv ^= (_data[i] << (i*8));
}
return rv;
}
@Override
@ -100,4 +106,4 @@ public class PrivateKey extends DataStructureImpl {
return KeyGenerator.getPublicKey(this);
}
}
}

10
core/java/src/net/i2p/data/PublicKey.java
View File

@ -72,9 +72,15 @@ public class PublicKey extends DataStructureImpl {
return DataHelper.eq(_data, ((PublicKey) obj)._data);
}
/** the key has enough randomness in it, use the first 4 bytes for speed */
@Override
public int hashCode() {
return DataHelper.hashCode(_data);
int rv = 0;
if (_data != null) {
for (int i = 0; i < 4; i++)
rv ^= (_data[i] << (i*8));
}
return rv;
}
@Override
@ -94,4 +100,4 @@ public class PublicKey extends DataStructureImpl {
return buf.toString();
}
}
}

6
core/java/src/net/i2p/data/RouterAddress.java
View File

@ -130,10 +130,10 @@ public class RouterAddress extends DataStructureImpl {
&& DataHelper.eq(getTransportStyle(), addr.getTransportStyle());
}
/** the style should be sufficient, for speed */
@Override
public int hashCode() {
return getCost() + DataHelper.hashCode(getTransportStyle()) + DataHelper.hashCode(getExpiration())
+ DataHelper.hashCode(getOptions());
return DataHelper.hashCode(getTransportStyle());
}
@Override
@ -152,4 +152,4 @@ public class RouterAddress extends DataStructureImpl {
buf.append("]");
return buf.toString();
}
}
}

8
core/java/src/net/i2p/data/RouterIdentity.java
View File

@ -101,10 +101,12 @@ public class RouterIdentity extends DataStructureImpl {
&& DataHelper.eq(getPublicKey(), ident.getPublicKey());
}
/** the public key has enough randomness in it to use it by itself for speed */
@Override
public int hashCode() {
return DataHelper.hashCode(getCertificate()) + DataHelper.hashCode(getSigningPublicKey())
+ DataHelper.hashCode(getPublicKey());
if (_publicKey == null)
return 0;
return _publicKey.hashCode();
}
@Override
@ -140,4 +142,4 @@ public class RouterIdentity extends DataStructureImpl {
__calculatedHash = SHA256Generator.getInstance().calculateHash(identBytes);
return __calculatedHash;
}
}
}

10
core/java/src/net/i2p/data/SessionKey.java
View File

@ -76,9 +76,15 @@ public class SessionKey extends DataStructureImpl {
return DataHelper.eq(_data, ((SessionKey) obj)._data);
}
/** the key has enough randomness in it, use the first 4 bytes for speed */
@Override
public int hashCode() {
return DataHelper.hashCode(_data);
int rv = 0;
if (_data != null) {
for (int i = 0; i < 4; i++)
rv ^= (_data[i] << (i*8));
}
return rv;
}
@Override
@ -98,4 +104,4 @@ public class SessionKey extends DataStructureImpl {
buf.append("]");
return buf.toString();
}
}
}

10
core/java/src/net/i2p/data/Signature.java
View File

@ -62,9 +62,15 @@ public class Signature extends DataStructureImpl {
return DataHelper.eq(_data, ((Signature) obj)._data);
}
/** the sig has enough randomness in it, use the first 4 bytes for speed */
@Override
public int hashCode() {
return DataHelper.hashCode(_data);
int rv = 0;
if (_data != null) {
for (int i = 0; i < 4; i++)
rv ^= (_data[i] << (i*8));
}
return rv;
}
@Override
@ -83,4 +89,4 @@ public class Signature extends DataStructureImpl {
buf.append("]");
return buf.toString();
}
}
}

10
core/java/src/net/i2p/data/SigningPrivateKey.java
View File

@ -68,9 +68,15 @@ public class SigningPrivateKey extends DataStructureImpl {
return DataHelper.eq(_data, ((SigningPrivateKey) obj)._data);
}
/** the key has enough randomness in it, use the first 4 bytes for speed */
@Override
public int hashCode() {
return DataHelper.hashCode(_data);
int rv = 0;
if (_data != null) {
for (int i = 0; i < 4; i++)
rv ^= (_data[i] << (i*8));
}
return rv;
}
@Override
@ -96,4 +102,4 @@ public class SigningPrivateKey extends DataStructureImpl {
public SigningPublicKey toPublic() {
return KeyGenerator.getSigningPublicKey(this);
}
}
}

10
core/java/src/net/i2p/data/SigningPublicKey.java
View File

@ -67,9 +67,15 @@ public class SigningPublicKey extends DataStructureImpl {
return DataHelper.eq(_data, ((SigningPublicKey) obj)._data);
}
/** the key has enough randomness in it, use the first 4 bytes for speed */
@Override
public int hashCode() {
return DataHelper.hashCode(_data);
int rv = 0;
if (_data != null) {
for (int i = 0; i < 4; i++)
rv ^= (_data[i] << (i*8));
}
return rv;
}
@Override
@ -88,4 +94,4 @@ public class SigningPublicKey extends DataStructureImpl {
buf.append("]");
return buf.toString();
}
}
}

68
core/java/src/net/i2p/util/DecayingBloomFilter.java
View File

@ -14,6 +14,10 @@ import org.xlattice.crypto.filters.BloomSHA1;
* entries per second with virtually no false positive rate. Down the line,
* this may be refactored to allow tighter control of the size necessary for the
* contained bloom filters, but a fixed 2MB overhead isn't that bad.
*
* NOTE: At 1MBps, the tunnel IVV will see an unacceptable false positive rate
* of almost 0.1% with the current m and k values; however using DHS instead will use 30MB.
* Further analysis and tweaking for the tunnel IVV may be required.
*/
public class DecayingBloomFilter {
private I2PAppContext _context;
@ -26,13 +30,18 @@ public class DecayingBloomFilter {
private byte _extended[];
private byte _longToEntry[];
private long _longToEntryMask;
private long _currentDuplicates;
protected long _currentDuplicates;
private boolean _keepDecaying;
private DecayEvent _decayEvent;
/** just for logging */
private String _name;
private static final int DEFAULT_M = 23;
private static final boolean ALWAYS_MISS = false;
/** noop for DHS */
public DecayingBloomFilter() {}
/**
* Create a bloom filter that will decay its entries over time.
*
@ -42,9 +51,15 @@ public class DecayingBloomFilter {
* against with sufficient random values.
*/
public DecayingBloomFilter(I2PAppContext context, int durationMs, int entryBytes) {
this(context, durationMs, entryBytes, "DBF");
}
/** @param name just for logging / debugging / stats */
public DecayingBloomFilter(I2PAppContext context, int durationMs, int entryBytes, String name) {
_context = context;
_log = context.logManager().getLog(DecayingBloomFilter.class);
_entryBytes = entryBytes;
_name = name;
// this is instantiated in four different places, they may have different
// requirements, but for now use this as a gross method of memory reduction.
// m == 23 => 1MB each BloomSHA1 (4 pairs = 8MB total)
@ -67,6 +82,17 @@ public class DecayingBloomFilter {
_decayEvent = new DecayEvent();
_keepDecaying = true;
SimpleTimer.getInstance().addEvent(_decayEvent, _durationMs);
if (_log.shouldLog(Log.WARN))
_log.warn("New DBF " + name + " m = " + m + " entryBytes = " + entryBytes +
" numExtenders = " + numExtenders + " cycle (s) = " + (durationMs / 1000));
// try to get a handle on memory usage vs. false positives
context.statManager().createRateStat("router.decayingBloomFilter." + name + ".size",
"Size", "Router", new long[] { Math.max(60*1000, durationMs) });
context.statManager().createRateStat("router.decayingBloomFilter." + name + ".dups",
"1000000 * Duplicates/Size", "Router", new long[] { Math.max(60*1000, durationMs) });
context.statManager().createRateStat("router.decayingBloomFilter." + name + ".log10(falsePos)",
"log10 of the false positive rate (must have net.i2p.util.DecayingBloomFilter=DEBUG)",
"Router", new long[] { Math.max(60*1000, durationMs) });
}
public long getCurrentDuplicateCount() { return _currentDuplicates; }
@ -196,9 +222,12 @@ public class DecayingBloomFilter {
private void decay() {
int currentCount = 0;
long dups = 0;
double fpr = 0d;
synchronized (this) {
BloomSHA1 tmp = _previous;
currentCount = _current.size();
if (_log.shouldLog(Log.DEBUG) && currentCount > 0)
fpr = _current.falsePositives();
_previous = _current;
_current = tmp;
_current.clear();
@ -206,8 +235,19 @@ public class DecayingBloomFilter {
_currentDuplicates = 0;
}
if (_log.shouldLog(Log.DEBUG))
_log.debug("Decaying the filter after inserting " + currentCount
+ " elements and " + dups + " false positives");
_log.debug("Decaying the filter " + _name + " after inserting " + currentCount
+ " elements and " + dups + " false positives with FPR = " + fpr);
_context.statManager().addRateData("router.decayingBloomFilter." + _name + ".size",
currentCount, 0);
if (currentCount > 0)
_context.statManager().addRateData("router.decayingBloomFilter." + _name + ".dups",
1000l*1000*dups/currentCount, 0);
if (fpr > 0d) {
// only if log.shouldLog(Log.DEBUG) ...
long exponent = (long) Math.log10(fpr);
_context.statManager().addRateData("router.decayingBloomFilter." + _name + ".log10(falsePos)",
exponent, 0);
}
}
private class DecayEvent implements SimpleTimer.TimedEvent {
@ -219,18 +259,27 @@ public class DecayingBloomFilter {
}
}
/**
* Theoretical false positive rate for 16 KBps: 1.17E-21
* Theoretical false positive rate for 24 KBps: 9.81E-20
* Theoretical false positive rate for 32 KBps: 2.24E-18
* Theoretical false positive rate for 256 KBps: 7.45E-9
* Theoretical false positive rate for 512 KBps: 5.32E-6
* Theoretical false positive rate for 1024 KBps: 1.48E-3
*/
public static void main(String args[]) {
int kbps = 256;
int iterations = 100;
int iterations = 10;
testByLong(kbps, iterations);
testByBytes(kbps, iterations);
}
public static void testByLong(int kbps, int numRuns) {
private static void testByLong(int kbps, int numRuns) {
int messages = 60 * 10 * kbps;
Random r = new Random();
DecayingBloomFilter filter = new DecayingBloomFilter(I2PAppContext.getGlobalContext(), 600*1000, 8);
int falsePositives = 0;
long totalTime = 0;
double fpr = 0d;
for (int j = 0; j < numRuns; j++) {
long start = System.currentTimeMillis();
for (int i = 0; i < messages; i++) {
@ -240,15 +289,17 @@ public class DecayingBloomFilter {
}
}
totalTime += System.currentTimeMillis() - start;
fpr = filter.getFalsePositiveRate();
filter.clear();
}
filter.stopDecaying();
System.out.println("False postive rate should be " + fpr);
System.out.println("After " + numRuns + " runs pushing " + messages + " entries in "
+ DataHelper.formatDuration(totalTime/numRuns) + " per run, there were "
+ falsePositives + " false positives");
}
public static void testByBytes(int kbps, int numRuns) {
private static void testByBytes(int kbps, int numRuns) {
byte iv[][] = new byte[60*10*kbps][16];
Random r = new Random();
for (int i = 0; i < iv.length; i++)
@ -257,18 +308,21 @@ public class DecayingBloomFilter {
DecayingBloomFilter filter = new DecayingBloomFilter(I2PAppContext.getGlobalContext(), 600*1000, 16);
int falsePositives = 0;
long totalTime = 0;
double fpr = 0d;
for (int j = 0; j < numRuns; j++) {
long start = System.currentTimeMillis();
for (int i = 0; i < iv.length; i++) {
if (filter.add(iv[i])) {
falsePositives++;
System.out.println("False positive " + falsePositives + " (testByLong j=" + j + " i=" + i + ")");
System.out.println("False positive " + falsePositives + " (testByBytes j=" + j + " i=" + i + ")");
}
}
totalTime += System.currentTimeMillis() - start;
fpr = filter.getFalsePositiveRate();
filter.clear();
}
filter.stopDecaying();
System.out.println("False postive rate should be " + fpr);
System.out.println("After " + numRuns + " runs pushing " + iv.length + " entries in "
+ DataHelper.formatDuration(totalTime/numRuns) + " per run, there were "
+ falsePositives + " false positives");

380
core/java/src/net/i2p/util/DecayingHashSet.java Normal file
View File

@ -0,0 +1,380 @@
package net.i2p.util;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.Random;
import net.i2p.I2PAppContext;
import net.i2p.data.DataHelper;
/**
* Double buffered hash set.
* Since DecayingBloomFilter was instantiated 4 times for a total memory usage
* of 8MB, it seemed like we could do a lot better, given these usage stats
* on a class L router:
*
* ./router/java/src/net/i2p/router/tunnel/BuildMessageProcessor.java:
* 32 bytes, peak 10 entries in 1m
*
* ./router/java/src/net/i2p/router/transport/udp/InboundMessageFragments.java:
* 4 bytes, peak 150 entries in 10s
*
* ./router/java/src/net/i2p/router/MessageValidator.java:
* 8 bytes, peak 1K entries in 2m
*
* ./router/java/src/net/i2p/router/tunnel/BloomFilterIVValidator.java:
* 16 bytes, peak 15K entries in 10m
*
* If the ArrayWrapper object in the HashSet is 50 bytes, and BloomSHA1(23, 11) is 1MB,
* then for less than 20K entries this is smaller.
* And this uses space proportional to traffiic, so it doesn't penalize small routers
* with a fixed 8MB.
* So let's try it for the first 2 or 3, for now.
*
* Also, DBF is syncrhonized, and uses SimpleTimer.
* Here we use a read/write lock, with synchronization only
* when switching double buffers, and we use SimpleScheduler.
*
* Yes, we could stare at stats all day, and try to calculate an acceptable
* false-positive rate for each of the above uses, then estimate the DBF size
* required to meet that rate for a given usage. Or even start adjusting the
* Bloom filter m and k values on a per-DBF basis. But it's a whole lot easier
* to implement something with a zero false positive rate, and uses less memory
* for almost all bandwidth classes.
*
* This has a strictly zero false positive rate for <= 8 byte keys.
* For larger keys, it is 1 / (2**64) ~= 5E-20, which is better than
* DBF for any entry count greater than about 14K.
*
* DBF has a zero false negative rate over the period
* 2 * durationMs. And a 100% false negative rate beyond that period.
* This has the same properties.
*
* This performs about twice as fast as DBF in the test below.
*
* @author zzz
*/
public class DecayingHashSet extends DecayingBloomFilter {
private final I2PAppContext _context;
private final Log _log;
private ConcurrentHashSet<ArrayWrapper> _current;
private ConcurrentHashSet<ArrayWrapper> _previous;
private int _durationMs;
private int _entryBytes;
private volatile boolean _keepDecaying;
private final DecayEvent _decayEvent;
/** just for logging */
private final String _name;
/** synchronize against this lock when switching double buffers */
private final ReentrantReadWriteLock _reorganizeLock = new ReentrantReadWriteLock(true);
/**
* Create a double-buffered hash set that will decay its entries over time.
*
* @param durationMs entries last for at least this long, but no more than twice this long
* @param entryBytes how large are the entries to be added? 1 to 32 bytes
*/
public DecayingHashSet(I2PAppContext context, int durationMs, int entryBytes) {
this(context, durationMs, entryBytes, "DHS");
}
/** @param name just for logging / debugging / stats */
public DecayingHashSet(I2PAppContext context, int durationMs, int entryBytes, String name) {
if (entryBytes <= 0 || entryBytes > 32)
throw new IllegalArgumentException("Bad size");
_context = context;
_log = context.logManager().getLog(DecayingHashSet.class);
_entryBytes = entryBytes;
_name = name;
_current = new ConcurrentHashSet(128);
_previous = new ConcurrentHashSet(128);
_durationMs = durationMs;
_currentDuplicates = 0;
_decayEvent = new DecayEvent();
_keepDecaying = true;
SimpleScheduler.getInstance().addEvent(_decayEvent, _durationMs);
if (_log.shouldLog(Log.WARN))
_log.warn("New DHS " + name + " entryBytes = " + entryBytes +
" cycle (s) = " + (durationMs / 1000));
// try to get a handle on memory usage vs. false positives
context.statManager().createRateStat("router.decayingHashSet." + name + ".size",
"Size", "Router", new long[] { Math.max(60*1000, durationMs) });
context.statManager().createRateStat("router.decayingHashSet." + name + ".dups",
"1000000 * Duplicates/Size", "Router", new long[] { Math.max(60*1000, durationMs) });
}
/** unsynchronized but only used for logging elsewhere */
@Override
public int getInsertedCount() {
return _current.size() + _previous.size();
}
/** pointless, only used for logging elsewhere */
@Override
public double getFalsePositiveRate() {
if (_entryBytes <= 8)
return 0d;
return 1d / Math.pow(2d, 64d); // 5.4E-20
}
/**
* @return true if the entry added is a duplicate
*
*/
@Override
public boolean add(byte entry[], int off, int len) {
if (entry == null)
throw new IllegalArgumentException("Null entry");
if (len != _entryBytes)
throw new IllegalArgumentException("Bad entry [" + len + ", expected "
+ _entryBytes + "]");
getReadLock();
try {
return locked_add(entry, off, len, true);
} finally { releaseReadLock(); }
}
/**
* @return true if the entry added is a duplicate. the number of low order
* bits used is determined by the entryBytes parameter used on creation of the
* filter.
*
*/
@Override
public boolean add(long entry) {
return add(entry, true);
}
/**
* @return true if the entry is already known. this does NOT add the
* entry however.
*
*/
@Override
public boolean isKnown(long entry) {
return add(entry, false);
}
private boolean add(long entry, boolean addIfNew) {
int len = Math.min(8, _entryBytes);
byte[] b = toLong(len, entry);
getReadLock();
try {
return locked_add(b, 0, len, addIfNew);
} finally { releaseReadLock(); }
}
/** from DataHelper, except negative values ok */
private static byte[] toLong(int numBytes, long value) {
byte target[] = new byte[numBytes];
for (int i = 0; i < numBytes; i++)
target[numBytes-i-1] = (byte)(value >>> (i*8));
return target;
}
/** so many questions... */
private boolean locked_add(byte entry[], int offset, int len, boolean addIfNew) {
ArrayWrapper w = new ArrayWrapper(entry, offset, len);
boolean seen = _current.contains(w);
seen = seen || _previous.contains(w);
if (seen) {
// why increment if addIfNew == false?
// why not add to current if only in previous?
_currentDuplicates++;
} else if (addIfNew) {
_current.add(w);
// why add to previous?
_previous.add(w);
}
return seen;
}
@Override
public void clear() {
_current.clear();
_previous.clear();
_currentDuplicates = 0;
}
/** super doesn't call clear, but neither do the users, so it seems like we should here */
@Override
public void stopDecaying() {
_keepDecaying = false;
clear();
}
private void decay() {
int currentCount = 0;
long dups = 0;
if (!getWriteLock())
return;
try {
ConcurrentHashSet<ArrayWrapper> tmp = _previous;
currentCount = _current.size();
_previous = _current;
_current = tmp;
_current.clear();
dups = _currentDuplicates;
_currentDuplicates = 0;
} finally { releaseWriteLock(); }
if (_log.shouldLog(Log.DEBUG))
_log.debug("Decaying the filter " + _name + " after inserting " + currentCount
+ " elements and " + dups + " false positives");
_context.statManager().addRateData("router.decayingHashSet." + _name + ".size",
currentCount, 0);
if (currentCount > 0)
_context.statManager().addRateData("router.decayingHashSet." + _name + ".dups",
1000l*1000*dups/currentCount, 0);
}
/** if decay() ever blows up, we won't reschedule, and will grow unbounded, but it seems unlikely */
private class DecayEvent implements SimpleTimer.TimedEvent {
public void timeReached() {
if (_keepDecaying) {
decay();
SimpleScheduler.getInstance().addEvent(DecayEvent.this, _durationMs);
}
}
}
private void getReadLock() {
_reorganizeLock.readLock().lock();
}
private void releaseReadLock() {
_reorganizeLock.readLock().unlock();
}
/** @return true if the lock was acquired */
private boolean getWriteLock() {
try {
boolean rv = _reorganizeLock.writeLock().tryLock(5000, TimeUnit.MILLISECONDS);
if (!rv)
_log.error("no lock, size is: " + _reorganizeLock.getQueueLength(), new Exception("rats"));
return rv;
} catch (InterruptedException ie) {}
return false;
}
private void releaseWriteLock() {
_reorganizeLock.writeLock().unlock();
}
/**
* This saves the data as-is if the length is <= 8 bytes,
* otherwise it stores an 8-byte hash.
* Hash function is from DataHelper, modded to get
* the maximum entropy given the length of the data.
*/
private static class ArrayWrapper {
private long _longhashcode;
public ArrayWrapper(byte[] b, int offset, int len) {
int idx = offset;
int shift = Math.min(8, 64 / len);
for (int i = 0; i < len; i++) {
// xor better than + in tests
_longhashcode ^= (((long) b[idx++]) << (i * shift));
}
}
public int hashCode() {
return (int) _longhashcode;
}
public long longHashCode() {
return _longhashcode;
}
public boolean equals(Object o) {
if (o == null || !(o instanceof ArrayWrapper))
return false;
return ((ArrayWrapper) o).longHashCode() == _longhashcode;
}
}
/**
* vs. DBF, this measures 1.93x faster for testByLong and 2.46x faster for testByBytes.
*/
public static void main(String args[]) {
/** KBytes per sec, 1 message per KByte */
int kbps = 256;
int iterations = 10;
//testSize();
testByLong(kbps, iterations);
testByBytes(kbps, iterations);
}
/** and the answer is: 49.9 bytes. The ArrayWrapper alone measured 16, so that's 34 for the HashSet entry. */
/*****
private static void testSize() {
int qty = 256*1024;
byte b[] = new byte[8];
Random r = new Random();
long old = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
ConcurrentHashSet foo = new ConcurrentHashSet(qty);
for (int i = 0; i < qty; i++) {
r.nextBytes(b);
foo.add(new ArrayWrapper(b, 0, 8));
}
long used = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
System.out.println("Memory per ArrayWrapper: " + (((double) (used - old)) / qty));
}
*****/
/** 8 bytes, simulate the router message validator */
private static void testByLong(int kbps, int numRuns) {
int messages = 60 * 10 * kbps;
Random r = new Random();
DecayingBloomFilter filter = new DecayingHashSet(I2PAppContext.getGlobalContext(), 600*1000, 8);
int falsePositives = 0;
long totalTime = 0;
for (int j = 0; j < numRuns; j++) {
long start = System.currentTimeMillis();
for (int i = 0; i < messages; i++) {
if (filter.add(r.nextLong())) {
falsePositives++;
System.out.println("False positive " + falsePositives + " (testByLong j=" + j + " i=" + i + ")");
}
}
totalTime += System.currentTimeMillis() - start;
filter.clear();
}
System.out.println("False postive rate should be " + filter.getFalsePositiveRate());
filter.stopDecaying();
System.out.println("After " + numRuns + " runs pushing " + messages + " entries in "
+ DataHelper.formatDuration(totalTime/numRuns) + " per run, there were "
+ falsePositives + " false positives");
}
/** 16 bytes, simulate the tunnel IV validator */
private static void testByBytes(int kbps, int numRuns) {
byte iv[][] = new byte[60*10*kbps][16];
Random r = new Random();
for (int i = 0; i < iv.length; i++)
r.nextBytes(iv[i]);
DecayingBloomFilter filter = new DecayingHashSet(I2PAppContext.getGlobalContext(), 600*1000, 16);
int falsePositives = 0;
long totalTime = 0;
for (int j = 0; j < numRuns; j++) {
long start = System.currentTimeMillis();
for (int i = 0; i < iv.length; i++) {
if (filter.add(iv[i])) {
falsePositives++;
System.out.println("False positive " + falsePositives + " (testByBytes j=" + j + " i=" + i + ")");
}
}
totalTime += System.currentTimeMillis() - start;
filter.clear();
}
System.out.println("False postive rate should be " + filter.getFalsePositiveRate());
filter.stopDecaying();
System.out.println("After " + numRuns + " runs pushing " + iv.length + " entries in "
+ DataHelper.formatDuration(totalTime/numRuns) + " per run, there were "
+ falsePositives + " false positives");
}
}

48
core/java/src/net/i2p/util/EepGet.java
View File

@ -27,7 +27,7 @@ import net.i2p.data.DataHelper;
* Bug: a malformed url http://example.i2p (no trailing '/') fails cryptically
*/
public class EepGet {
private I2PAppContext _context;
protected I2PAppContext _context;
protected Log _log;
protected boolean _shouldProxy;
private String _proxyHost;
@ -35,8 +35,8 @@ public class EepGet {
protected int _numRetries;
private long _minSize; // minimum and maximum acceptable response size, -1 signifies unlimited,
private long _maxSize; // applied both against whole responses and chunks
private String _outputFile;
private OutputStream _outputStream;
protected String _outputFile;
protected OutputStream _outputStream;
/** url we were asked to fetch */
protected String _url;
/** the URL we actually fetch from (may differ from the _url in case of redirect) */
@ -47,10 +47,10 @@ public class EepGet {
private boolean _keepFetching;
private Socket _proxy;
private OutputStream _proxyOut;
private InputStream _proxyIn;
protected OutputStream _proxyOut;
protected InputStream _proxyIn;
protected OutputStream _out;
private long _alreadyTransferred;
protected long _alreadyTransferred;
private long _bytesTransferred;
protected long _bytesRemaining;
protected int _currentAttempt;
@ -67,6 +67,10 @@ public class EepGet {
protected long _fetchInactivityTimeout;
protected int _redirects;
protected String _redirectLocation;
/** this will be replaced by the HTTP Proxy if we are using it */
protected static final String USER_AGENT = "Wget/1.11.4";
protected static final long CONNECT_TIMEOUT = 45*1000;
protected static final long INACTIVITY_TIMEOUT = 60*1000;
public EepGet(I2PAppContext ctx, String proxyHost, int proxyPort, int numRetries, String outputFile, String url) {
this(ctx, true, proxyHost, proxyPort, numRetries, outputFile, url);
@ -118,7 +122,7 @@ public class EepGet {
_transferFailed = false;
_headersRead = false;
_aborted = false;
_fetchHeaderTimeout = 45*1000;
_fetchHeaderTimeout = CONNECT_TIMEOUT;
_listeners = new ArrayList(1);
_etag = etag;
_lastModified = lastModified;
@ -134,7 +138,7 @@ public class EepGet {
int numRetries = 5;
int markSize = 1024;
int lineLen = 40;
int inactivityTimeout = 60*1000;
long inactivityTimeout = INACTIVITY_TIMEOUT;
String etag = null;
String saveAs = null;
String url = null;
@ -183,7 +187,7 @@ public class EepGet {
EepGet get = new EepGet(I2PAppContext.getGlobalContext(), true, proxyHost, proxyPort, numRetries, saveAs, url, true, etag);
get.addStatusListener(get.new CLIStatusListener(markSize, lineLen));
get.fetch(45*1000, -1, inactivityTimeout);
get.fetch(CONNECT_TIMEOUT, -1, inactivityTimeout);
}
public static String suggestName(String url) {
@ -216,7 +220,7 @@ public class EepGet {
return buf.toString();
}
protected static void usage() {
private static void usage() {
System.err.println("EepGet [-p 127.0.0.1:4444] [-n #retries] [-o outputFile] [-m markSize lineLen] [-t timeout] url");
}
@ -247,7 +251,7 @@ public class EepGet {
public void headerReceived(String url, int currentAttempt, String key, String val);
public void attempting(String url);
}
private class CLIStatusListener implements StatusListener {
protected class CLIStatusListener implements StatusListener {
private int _markSize;
private int _lineSize;
private long _startedOn;
@ -497,7 +501,7 @@ public class EepGet {
if (_fetchInactivityTimeout > 0)
timeout.setInactivityTimeout(_fetchInactivityTimeout);
else
timeout.setInactivityTimeout(60*1000);
timeout.setInactivityTimeout(INACTIVITY_TIMEOUT);
if (_redirectLocation != null) {
try {
@ -829,12 +833,12 @@ public class EepGet {
}
}
private void increment(byte[] lookahead, int cur) {
private static void increment(byte[] lookahead, int cur) {
lookahead[0] = lookahead[1];
lookahead[1] = lookahead[2];
lookahead[2] = (byte)cur;
}
private boolean isEndOfHeaders(byte lookahead[]) {
private static boolean isEndOfHeaders(byte lookahead[]) {
byte first = lookahead[0];
byte second = lookahead[1];
byte third = lookahead[2];
@ -844,7 +848,7 @@ public class EepGet {
/** we ignore any potential \r, since we trim it on write anyway */
private static final byte NL = '\n';
private boolean isNL(byte b) { return (b == NL); }
private static boolean isNL(byte b) { return (b == NL); }
protected void sendRequest(SocketTimeout timeout) throws IOException {
if (_outputStream != null) {
@ -895,7 +899,7 @@ public class EepGet {
}
protected String getRequest() throws IOException {
StringBuilder buf = new StringBuilder(512);
StringBuilder buf = new StringBuilder(2048);
boolean post = false;
if ( (_postData != null) && (_postData.length() > 0) )
post = true;
@ -906,7 +910,7 @@ public class EepGet {
String path = url.getPath();
String query = url.getQuery();
if (query != null)
path = path + "?" + query;
path = path + '?' + query;
if (!path.startsWith("/"))
path = "/" + path;
if ( (port == 80) || (port == 443) || (port <= 0) ) path = proto + "://" + host + path;
@ -923,12 +927,11 @@ public class EepGet {
buf.append(_alreadyTransferred);
buf.append("-\r\n");
}
buf.append("Accept-Encoding: \r\n");
if (_shouldProxy)
buf.append("X-Accept-Encoding: x-i2p-gzip;q=1.0, identity;q=0.5, deflate;q=0, gzip;q=0, *;q=0\r\n");
if (!_allowCaching) {
buf.append("Cache-control: no-cache\r\n");
buf.append("Pragma: no-cache\r\n");
buf.append("Cache-control: no-cache\r\n" +
"Pragma: no-cache\r\n");
}
if ((_etag != null) && (_alreadyTransferred <= 0)) {
buf.append("If-None-Match: ");
@ -942,7 +945,10 @@ public class EepGet {
}
if (post)
buf.append("Content-length: ").append(_postData.length()).append("\r\n");
buf.append("Connection: close\r\n\r\n");
// This will be replaced if we are going through I2PTunnelHTTPClient
buf.append("User-Agent: " + USER_AGENT + "\r\n" +
"Accept-Encoding: \r\n" +
"Connection: close\r\n\r\n");
if (post)
buf.append(_postData);
if (_log.shouldLog(Log.DEBUG))

2
core/java/src/net/i2p/util/EepGetScheduler.java
View File

@ -7,7 +7,7 @@ import java.util.List;
import net.i2p.I2PAppContext;
/**
*
* @deprecated unused a webapp version would be nice though
*/
public class EepGetScheduler implements EepGet.StatusListener {
private I2PAppContext _context;

4
core/java/src/net/i2p/util/EepHead.java
View File

@ -93,7 +93,7 @@ public class EepHead extends EepGet {
}
}
protected static void usage() {
private static void usage() {
System.err.println("EepHead [-p 127.0.0.1:4444] [-n #retries] [-t timeout] url");
}
@ -191,6 +191,8 @@ public class EepHead extends EepGet {
buf.append("Accept-Encoding: \r\n");
if (_shouldProxy)
buf.append("X-Accept-Encoding: x-i2p-gzip;q=1.0, identity;q=0.5, deflate;q=0, gzip;q=0, *;q=0\r\n");
// This will be replaced if we are going through I2PTunnelHTTPClient
buf.append("User-Agent: " + USER_AGENT + "\r\n");
buf.append("Connection: close\r\n\r\n");
if (_log.shouldLog(Log.DEBUG))
_log.debug("Request: [" + buf.toString() + "]");

11
core/java/src/net/i2p/util/SocketTimeout.java
View File

@ -5,6 +5,15 @@ import java.net.Socket;
import java.text.SimpleDateFormat;
import java.util.Date;
/**
* This should be deprecated.
* It is only used by EepGet, and it uses the inefficient SimpleTimer.
* The only advantage seems to be a total timeout period, which is the second
* argument to EepGet.fetch(headerTimeout, totalTimeout, inactivityTimeout),
* which is most likely always set to -1.
*
* Use socket.setsotimeout instead?
*/
public class SocketTimeout implements SimpleTimer.TimedEvent {
private Socket _targetSocket;
private long _startTime;
@ -69,4 +78,4 @@ public class SocketTimeout implements SimpleTimer.TimedEvent {
buf.append("cancelled? ").append(_cancelled);
return buf.toString();
}
}
}