Changeset 48a055d


Ignore:
Timestamp:
Mar 9, 2017 7:40:17 PM (3 years ago)
Author:
zzz <zzz@…>
Branches:
master
Children:
072d990
Parents:
9146f3c
Message:

Streaming: Don't change buffer size when max message size is adjusted
Only change max message size when buffer is empty

Files:
3 edited

Legend:

Unmodified
Added
Removed
  • apps/streaming/java/src/net/i2p/client/streaming/impl/MessageOutputStream.java

    r9146f3c r48a055d  
    3131    private long _written;
    3232    private int _writeTimeout;
    33     private ByteCache _dataCache;
     33    private final ByteCache _dataCache;
     34    private final int _originalBufferSize;
     35    private int _currentBufferSize;
    3436    private final Flusher _flusher;
    3537    private volatile long _lastBuffered;
     
    6971        super();
    7072        _dataCache = ByteCache.getInstance(128, bufSize);
     73        _originalBufferSize = bufSize;
     74        _currentBufferSize = bufSize;
    7175        _context = ctx;
    7276        _log = ctx.logManager().getLog(MessageOutputStream.class);
     
    7680        _writeTimeout = -1;
    7781        _passiveFlushDelay = passiveFlushDelay;
    78         _nextBufferSize = -1;
     82        _nextBufferSize = 0;
    7983        //_sendPeriodBeginTime = ctx.clock().now();
    8084        //_context.statManager().createRateStat("stream.sendBps", "How fast we pump data through the stream", "Stream", new long[] { 60*1000, 5*60*1000, 60*60*1000 });
     
    9397    public int getWriteTimeout() { return _writeTimeout; }
    9498
    95     public void setBufferSize(int size) { _nextBufferSize = size; }
     99    /**
     100     *  Caller should enforce a sane minimum.
     101     *
     102     *  @param size must be greater than 0, and smaller than or equal to bufSize in constructor
     103     */
     104    public void setBufferSize(int size) {
     105        if (size <= 0 || size > _originalBufferSize)
     106            return;
     107        _nextBufferSize = size;
     108    }
    96109   
    97110    @Override
     
    116129            // code that reads from it is synchronized
    117130            synchronized (_dataLock) {
     131                // To simplify the code, and avoid losing data from shrinking the max size,
     132                // we only update max size when current buffer is empty
     133                final int maxBuffer = (_valid == 0) ? locked_updateBufferSize() : _currentBufferSize;
    118134                if (_buf == null) throw new IOException("closed (buffer went away)");
    119                 if (_valid + remaining < _buf.length) {
     135                if (_valid + remaining < maxBuffer) {
    120136                    // simply buffer the data, no flush
    121137                    System.arraycopy(b, cur, _buf, _valid, remaining);
     
    132148                    // repeating until we've pushed all of the
    133149                    // data through
    134                     int toWrite = _buf.length - _valid;
     150                    int toWrite = maxBuffer - _valid;
    135151                    System.arraycopy(b, cur, _buf, _valid, toWrite);
    136152                    remaining -= toWrite;
    137153                    cur += toWrite;
    138                     _valid = _buf.length;
     154                    _valid = maxBuffer;
    139155                    if (_log.shouldLog(Log.INFO))
    140156                        _log.info("write() direct valid = " + _valid);
     
    143159                    _valid = 0;                       
    144160                    throwAnyError();
    145                    
    146                     locked_updateBufferSize();
    147161                }
    148162            }
     
    208222     * If the other side requested we shrink our buffer, do so.
    209223     *
    210      */
    211     private final void locked_updateBufferSize() {
     224     * @return the current buffer size
     225     */
     226    private final int locked_updateBufferSize() {
    212227        int size = _nextBufferSize;
    213228        if (size > 0) {
    214229            // update the buffer size to the requested amount
    215             _dataCache.release(new ByteArray(_buf));
    216             _dataCache = ByteCache.getInstance(128, size);
    217             ByteArray ba = _dataCache.acquire();
    218             _buf = ba.getData();
    219             _nextBufferSize = -1;
    220         }
     230            // No, never do this, to avoid ByteCache churn.
     231            //_dataCache.release(new ByteArray(_buf));
     232            //_dataCache = ByteCache.getInstance(128, size);
     233            //ByteArray ba = _dataCache.acquire();
     234            //_buf = ba.getData();
     235            _currentBufferSize = size;
     236            _nextBufferSize = 0;
     237        }
     238        return _currentBufferSize;
    221239    }
    222240   
     
    274292                        _written += _valid;
    275293                        _valid = 0;
    276                         locked_updateBufferSize();
    277294                        _dataLock.notifyAll();
    278295                        sent = true;
     
    337354                _written += _valid;
    338355                _valid = 0;
    339                 locked_updateBufferSize();
    340356                _dataLock.notifyAll();
    341357            }
     
    410426                _buf = null;
    411427                _valid = 0;
    412                 locked_updateBufferSize();
    413428            }
    414429            _dataLock.notifyAll();
     
    495510            _written += _valid;
    496511            _valid = 0;
    497             locked_updateBufferSize();
    498512            _dataLock.notifyAll();
    499513        }
  • history.txt

    r9146f3c r48a055d  
     12017-03-09 zzz
     2 * i2psnark: Prevent RuntimeException caused by corrupt i2psnark.dht.dat file
     3 * Router: Set default sig type to EdDSA for non-Android ARM
     4 * Streaming: Don't change buffer size when max message size is adjusted
     5
    162017-03-06 zzz
    27 * CPUID:
  • router/java/src/net/i2p/router/RouterVersion.java

    r9146f3c r48a055d  
    1919    public final static String ID = "Monotone";
    2020    public final static String VERSION = CoreVersion.VERSION;
    21     public final static long BUILD = 2;
     21    public final static long BUILD = 3;
    2222
    2323    /** for example "-test" */
Note: See TracChangeset for help on using the changeset viewer.