Changeset ab92206


Ignore:
Timestamp:
Jan 20, 2009 5:20:37 PM (12 years ago)
Author:
zzz <zzz@…>
Branches:
master
Children:
6be5494
Parents:
0e2a422
Message:
Location:
apps/streaming/java/src/net/i2p/client/streaming
Files:
1 added
3 edited

Legend:

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

    r0e2a422 rab92206  
    4646    private long _highestAckedThrough;
    4747    private boolean _isInbound;
     48    private boolean _updatedShareOpts;
    4849    /** Packet ID (Long) to PacketLocal for sent but unacked packets */
    4950    private Map _outboundPackets;
     
    121122        _resetSentOn = -1;
    122123        _isInbound = false;
     124        _updatedShareOpts = false;
    123125        _connectionEvent = new ConEvent();
    124126        _context.statManager().createRateStat("stream.con.windowSizeAtCongestion", "How large was our send window when we send a dup?", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
     
    587589        _remotePeerSet = true;
    588590        _remotePeer = peer;
     591        // now that we know who the other end is, get the rtt etc. from the cache
     592        _connectionManager.updateOptsFromShare(this);
    589593    }
    590594   
     
    710714    public long getCloseReceivedOn() { return _closeReceivedOn; }
    711715    public void setCloseReceivedOn(long when) { _closeReceivedOn = when; }
    712    
     716
     717    public void updateShareOpts() {
     718        if (_closeSentOn > 0 && !_updatedShareOpts) {
     719            _connectionManager.updateShareOpts(this);
     720            _updatedShareOpts = true;
     721        }
     722    }
    713723    public void incrementUnackedPacketsReceived() { _unackedPacketsReceived++; }
    714724    public int getUnackedPacketsReceived() { return _unackedPacketsReceived; }
     
    9991009     * Coordinate the resends of a given packet
    10001010     */
    1001     private class ResendPacketEvent implements SimpleTimer.TimedEvent {
     1011    public class ResendPacketEvent implements SimpleTimer.TimedEvent {
    10021012        private PacketLocal _packet;
    10031013        private long _nextSendTime;
     
    11051115                }
    11061116               
    1107                 if (numSends - 1 <= _options.getMaxResends()) {
    1108                     if (_log.shouldLog(Log.INFO))
    1109                         _log.info("Resend packet " + _packet + " time " + numSends +
    1110                                   " activeResends: " + _activeResends +
    1111                                   " (wsize "
    1112                                   + newWindowSize + " lifetime "
    1113                                   + (_context.clock().now() - _packet.getCreatedOn()) + "ms)");
    1114                     _outboundQueue.enqueue(_packet);
    1115                     _lastSendTime = _context.clock().now();
    1116                 }
    1117                
    1118                 // acked during resending (... or somethin')
    1119                 if ( (_packet.getAckTime() > 0) && (_packet.getNumSends() > 1) ) {
    1120                     _activeResends--;
    1121                     synchronized (_outboundPackets) {
    1122                         _outboundPackets.notifyAll();
    1123                     }
    1124                     return true;
    1125                 }
    1126                
    11271117                if (numSends - 1 > _options.getMaxResends()) {
    11281118                    if (_log.shouldLog(Log.DEBUG))
     
    11381128                    if ( (timeout > MAX_RESEND_DELAY) || (timeout <= 0) )
    11391129                        timeout = MAX_RESEND_DELAY;
     1130                    // set this before enqueue() as it passes it on to the router
     1131                    _nextSendTime = timeout + _context.clock().now();
     1132
     1133                    if (_log.shouldLog(Log.INFO))
     1134                        _log.info("Resend packet " + _packet + " time " + numSends +
     1135                                  " activeResends: " + _activeResends +
     1136                                  " (wsize "
     1137                                  + newWindowSize + " lifetime "
     1138                                  + (_context.clock().now() - _packet.getCreatedOn()) + "ms)");
     1139                    _outboundQueue.enqueue(_packet);
     1140                    _lastSendTime = _context.clock().now();
     1141
    11401142                    if (_log.shouldLog(Log.DEBUG))
    11411143                        _log.debug("Scheduling resend in " + timeout + "ms for " + _packet);
    11421144                    RetransmissionTimer.getInstance().addEvent(ResendPacketEvent.this, timeout);
    1143                     _nextSendTime = timeout + _context.clock().now();
    1144                 }
     1145                }
     1146               
     1147                // acked during resending (... or somethin')
     1148                if ( (_packet.getAckTime() > 0) && (_packet.getNumSends() > 1) ) {
     1149                    _activeResends--;
     1150                    synchronized (_outboundPackets) {
     1151                        _outboundPackets.notifyAll();
     1152                    }
     1153                    return true;
     1154                }
     1155
    11451156                return true;
    11461157            } else {
  • apps/streaming/java/src/net/i2p/client/streaming/ConnectionManager.java

    r0e2a422 rab92206  
    3131    private SchedulerChooser _schedulerChooser;
    3232    private ConnectionPacketHandler _conPacketHandler;
     33    private TCBShare _tcbShare;
    3334    /** Inbound stream ID (Long) to Connection map */
    3435    private Map _connectionByInboundId;
     
    5354        _schedulerChooser = new SchedulerChooser(context);
    5455        _conPacketHandler = new ConnectionPacketHandler(context);
     56        _tcbShare = new TCBShare(context);
    5557        _session = session;
    5658        session.setSessionListener(_messageHandler);
     
    128130    public Connection receiveConnection(Packet synPacket) {
    129131        Connection con = new Connection(_context, this, _schedulerChooser, _outboundQueue, _conPacketHandler, new ConnectionOptions(_defaultOptions));
     132        _tcbShare.updateOptsFromShare(con);
    130133        con.setInbound();
    131134        long receiveId = _context.random().nextLong(Packet.MAX_STREAM_ID-1)+1;
     
    278281    public I2PSession getSession() { return _session; }
    279282    public PacketQueue getPacketQueue() { return _outboundQueue; }
     283    public void updateOptsFromShare(Connection con) { _tcbShare.updateOptsFromShare(con); }
     284    public void updateShareOpts(Connection con) { _tcbShare.updateShareOpts(con); }
    280285   
    281286    /**
     
    293298            _connectionLock.notifyAll();
    294299        }
     300        _tcbShare.stop();
    295301    }
    296302   
  • apps/streaming/java/src/net/i2p/client/streaming/ConnectionPacketHandler.java

    r0e2a422 rab92206  
    214214        }
    215215       
     216        // update the TCB Cache now that we've processed the acks and updated our rtt etc.
     217        if (isNew && packet.isFlagSet(Packet.FLAG_CLOSE) && packet.isFlagSet(Packet.FLAG_SIGNATURE_INCLUDED))
     218            con.updateShareOpts();
     219
    216220        //if (choke)
    217221        //    con.fastRetransmit();
Note: See TracChangeset for help on using the changeset viewer.