Changeset 593d4dc


Ignore:
Timestamp:
Aug 26, 2009 10:22:47 PM (11 years ago)
Author:
zzz <zzz@…>
Branches:
master
Children:
1ecf437
Parents:
93d366fe
Message:
Files:
1 added
5 edited

Legend:

Unmodified
Added
Removed
  • core/java/src/net/i2p/util/DecayingBloomFilter.java

    r93d366fe r593d4dc  
    1515 * this may be refactored to allow tighter control of the size necessary for the
    1616 * contained bloom filters, but a fixed 2MB overhead isn't that bad.
     17 *
     18 * NOTE: At 1MBps, the tunnel IVV will see an unacceptable false positive rate
     19 * of almost 0.1% with the current m and k values; however using DHS instead will use 30MB.
     20 * Further analysis and tweaking for the tunnel IVV may be required.
    1721 */
    1822public class DecayingBloomFilter {
     
    2731    private byte _longToEntry[];
    2832    private long _longToEntryMask;
    29     private long _currentDuplicates;
     33    protected long _currentDuplicates;
    3034    private boolean _keepDecaying;
    3135    private DecayEvent _decayEvent;
     36    /** just for logging */
     37    private String _name;
    3238   
    3339    private static final int DEFAULT_M = 23;
    3440    private static final boolean ALWAYS_MISS = false;
    3541   
     42    /** noop for DHS */
     43    public DecayingBloomFilter() {}
     44
    3645    /**
    3746     * Create a bloom filter that will decay its entries over time. 
     
    4352     */
    4453    public DecayingBloomFilter(I2PAppContext context, int durationMs, int entryBytes) {
     54        this(context, durationMs, entryBytes, "DBF");
     55    }
     56
     57    /** @param name just for logging / debugging / stats */
     58    public DecayingBloomFilter(I2PAppContext context, int durationMs, int entryBytes, String name) {
    4559        _context = context;
    4660        _log = context.logManager().getLog(DecayingBloomFilter.class);
    4761        _entryBytes = entryBytes;
     62        _name = name;
    4863        // this is instantiated in four different places, they may have different
    4964        // requirements, but for now use this as a gross method of memory reduction.
     
    6883        _keepDecaying = true;
    6984        SimpleTimer.getInstance().addEvent(_decayEvent, _durationMs);
     85        if (_log.shouldLog(Log.WARN))
     86           _log.warn("New DBF " + name + " m = " + m + " entryBytes = " + entryBytes +
     87                     " numExtenders = " + numExtenders + " cycle (s) = " + (durationMs / 1000));
     88        // try to get a handle on memory usage vs. false positives
     89        context.statManager().createRateStat("router.decayingBloomFilter." + name + ".size",
     90             "Size", "Router", new long[] { Math.max(60*1000, durationMs) });
     91        context.statManager().createRateStat("router.decayingBloomFilter." + name + ".dups",
     92             "1000000 * Duplicates/Size", "Router", new long[] { Math.max(60*1000, durationMs) });
     93        context.statManager().createRateStat("router.decayingBloomFilter." + name + ".log10(falsePos)",
     94             "log10 of the false positive rate (must have net.i2p.util.DecayingBloomFilter=DEBUG)",
     95             "Router", new long[] { Math.max(60*1000, durationMs) });
    7096    }
    7197   
     
    197223        int currentCount = 0;
    198224        long dups = 0;
     225        double fpr = 0d;
    199226        synchronized (this) {
    200227            BloomSHA1 tmp = _previous;
    201228            currentCount = _current.size();
     229            if (_log.shouldLog(Log.DEBUG) && currentCount > 0)
     230                fpr = _current.falsePositives();
    202231            _previous = _current;
    203232            _current = tmp;
     
    207236        }
    208237        if (_log.shouldLog(Log.DEBUG))
    209             _log.debug("Decaying the filter after inserting " + currentCount
    210                        + " elements and " + dups + " false positives");
     238            _log.debug("Decaying the filter " + _name + " after inserting " + currentCount
     239                       + " elements and " + dups + " false positives with FPR = " + fpr);
     240        _context.statManager().addRateData("router.decayingBloomFilter." + _name + ".size",
     241                                           currentCount, 0);
     242        if (currentCount > 0)
     243            _context.statManager().addRateData("router.decayingBloomFilter." + _name + ".dups",
     244                                               1000l*1000*dups/currentCount, 0);
     245        if (fpr > 0d) {
     246            // only if log.shouldLog(Log.DEBUG) ...
     247            long exponent = (long) Math.log10(fpr);
     248            _context.statManager().addRateData("router.decayingBloomFilter." + _name + ".log10(falsePos)",
     249                                               exponent, 0);
     250        }
    211251    }
    212252   
     
    220260    }
    221261   
     262    /**
     263     *  Theoretical false positive rate for   16 KBps: 1.17E-21
     264     *  Theoretical false positive rate for   24 KBps: 9.81E-20
     265     *  Theoretical false positive rate for   32 KBps: 2.24E-18
     266     *  Theoretical false positive rate for  256 KBps: 7.45E-9
     267     *  Theoretical false positive rate for  512 KBps: 5.32E-6
     268     *  Theoretical false positive rate for 1024 KBps: 1.48E-3
     269     */
    222270    public static void main(String args[]) {
    223271        int kbps = 256;
    224         int iterations = 100;
     272        int iterations = 10;
    225273        testByLong(kbps, iterations);
    226274        testByBytes(kbps, iterations);
    227275    }
    228     public static void testByLong(int kbps, int numRuns) {
     276    private static void testByLong(int kbps, int numRuns) {
    229277        int messages = 60 * 10 * kbps;
    230278        Random r = new Random();
     
    232280        int falsePositives = 0;
    233281        long totalTime = 0;
     282        double fpr = 0d;
    234283        for (int j = 0; j < numRuns; j++) {
    235284            long start = System.currentTimeMillis();
     
    241290            }
    242291            totalTime += System.currentTimeMillis() - start;
     292            fpr = filter.getFalsePositiveRate();
    243293            filter.clear();
    244294        }
    245295        filter.stopDecaying();
     296        System.out.println("False postive rate should be " + fpr);
    246297        System.out.println("After " + numRuns + " runs pushing " + messages + " entries in "
    247298                           + DataHelper.formatDuration(totalTime/numRuns) + " per run, there were "
     
    249300
    250301    }
    251     public static void testByBytes(int kbps, int numRuns) {
     302    private static void testByBytes(int kbps, int numRuns) {
    252303        byte iv[][] = new byte[60*10*kbps][16];
    253304        Random r = new Random();
     
    258309        int falsePositives = 0;
    259310        long totalTime = 0;
     311        double fpr = 0d;
    260312        for (int j = 0; j < numRuns; j++) {
    261313            long start = System.currentTimeMillis();
     
    263315                if (filter.add(iv[i])) {
    264316                    falsePositives++;
    265                     System.out.println("False positive " + falsePositives + " (testByLong j=" + j + " i=" + i + ")");
     317                    System.out.println("False positive " + falsePositives + " (testByBytes j=" + j + " i=" + i + ")");
    266318                }
    267319            }
    268320            totalTime += System.currentTimeMillis() - start;
     321            fpr = filter.getFalsePositiveRate();
    269322            filter.clear();
    270323        }
    271324        filter.stopDecaying();
     325        System.out.println("False postive rate should be " + fpr);
    272326        System.out.println("After " + numRuns + " runs pushing " + iv.length + " entries in "
    273327                           + DataHelper.formatDuration(totalTime/numRuns) + " per run, there were "
  • router/java/src/net/i2p/router/MessageValidator.java

    r93d366fe r593d4dc  
    22
    33import net.i2p.util.DecayingBloomFilter;
     4import net.i2p.util.DecayingHashSet;
    45import net.i2p.util.Log;
    56
     
    9697   
    9798    public void startup() {
    98         _filter = new DecayingBloomFilter(_context, (int)Router.CLOCK_FUDGE_FACTOR * 2, 8);
     99        _filter = new DecayingHashSet(_context, (int)Router.CLOCK_FUDGE_FACTOR * 2, 8, "RouterMV");
    99100    }
    100101   
  • router/java/src/net/i2p/router/transport/udp/InboundMessageFragments.java

    r93d366fe r593d4dc  
    66import net.i2p.router.RouterContext;
    77import net.i2p.util.DecayingBloomFilter;
     8import net.i2p.util.DecayingHashSet;
    89import net.i2p.util.Log;
    910
     
    5354        // array size (currently its tuned for 10 minute rates for the
    5455        // messageValidator)
    55         _recentlyCompletedMessages = new DecayingBloomFilter(_context, DECAY_PERIOD, 4);
     56        _recentlyCompletedMessages = new DecayingHashSet(_context, DECAY_PERIOD, 4, "UDPIMF");
    5657        _ackSender.startup();
    5758        _messageReceiver.startup();
  • router/java/src/net/i2p/router/tunnel/BloomFilterIVValidator.java

    r93d366fe r593d4dc  
    11package net.i2p.router.tunnel;
    22
    3 import net.i2p.I2PAppContext;
    43import net.i2p.data.ByteArray;
    54import net.i2p.data.DataHelper;
     5import net.i2p.router.RouterContext;
    66import net.i2p.util.ByteCache;
    77import net.i2p.util.DecayingBloomFilter;
     8import net.i2p.util.DecayingHashSet;
    89
    910/**
     
    1314 */
    1415public class BloomFilterIVValidator implements IVValidator {
    15     private I2PAppContext _context;
     16    private RouterContext _context;
    1617    private DecayingBloomFilter _filter;
    1718    private ByteCache _ivXorCache = ByteCache.getInstance(32, HopProcessor.IV_LENGTH);
     
    2425     */
    2526    private static final int HALFLIFE_MS = 10*60*1000;
    26     public BloomFilterIVValidator(I2PAppContext ctx, int KBps) {
     27    private static final int MIN_SHARE_KBPS_TO_USE_BLOOM = 64;
     28
     29    public BloomFilterIVValidator(RouterContext ctx, int KBps) {
    2730        _context = ctx;
    28         _filter = new DecayingBloomFilter(ctx, HALFLIFE_MS, 16);
     31        // Select the filter based on share bandwidth.
     32        // Note that at rates approaching 1MB, we need to do something else,
     33        // as the Bloom filter false positive rates approach 0.1%. FIXME
     34        if (getShareBandwidth(ctx) < MIN_SHARE_KBPS_TO_USE_BLOOM)
     35            _filter = new DecayingHashSet(ctx, HALFLIFE_MS, 16, "TunnelIVV"); // appx. 4MB max
     36        else
     37            _filter = new DecayingBloomFilter(ctx, HALFLIFE_MS, 16, "TunnelIVV");  // 2MB fixed
    2938        ctx.statManager().createRateStat("tunnel.duplicateIV", "Note that a duplicate IV was received", "Tunnels",
    3039                                         new long[] { 10*60*1000l, 60*60*1000l, 3*60*60*1000l, 24*60*60*1000l });
     
    4049    }
    4150    public void destroy() { _filter.stopDecaying(); }
     51
     52    private static int getShareBandwidth(RouterContext ctx) {
     53        int irateKBps = ctx.bandwidthLimiter().getInboundKBytesPerSecond();
     54        int orateKBps = ctx.bandwidthLimiter().getOutboundKBytesPerSecond();
     55        double pct = ctx.router().getSharePercentage();
     56        return (int) (pct * Math.min(irateKBps, orateKBps));
     57    }
    4258}
  • router/java/src/net/i2p/router/tunnel/BuildMessageProcessor.java

    r93d366fe r593d4dc  
    1111import net.i2p.data.i2np.TunnelBuildMessage;
    1212import net.i2p.util.DecayingBloomFilter;
     13import net.i2p.util.DecayingHashSet;
    1314import net.i2p.util.Log;
    1415
     
    2324   
    2425    public BuildMessageProcessor(I2PAppContext ctx) {
    25         _filter = new DecayingBloomFilter(ctx, 60*1000, 32);
     26        _filter = new DecayingHashSet(ctx, 60*1000, 32, "TunnelBMP");
    2627        ctx.statManager().createRateStat("tunnel.buildRequestDup", "How frequently we get dup build request messages", "Tunnels", new long[] { 60*1000, 10*60*1000 });
    2728    }
Note: See TracChangeset for help on using the changeset viewer.