diff --git a/udt/congestion.go b/udt/congestion.go index 18aed4c..3914d6e 100644 --- a/udt/congestion.go +++ b/udt/congestion.go @@ -37,9 +37,6 @@ type CongestionControlParms interface { // SetACKPerid sets the time between ACKs sent to the peer SetACKPeriod(time.Duration) - - // SetRTOPeriod overrides the default EXP timeout calculations waiting for data from the peer - SetRTOPeriod(time.Duration) } // CongestionControl controls how timing is handled and UDT connections tuned diff --git a/udt/udt.go b/udt/udt.go index 05a44aa..c8bd2d9 100644 --- a/udt/udt.go +++ b/udt/udt.go @@ -31,7 +31,6 @@ const ( TerminateReasonInvalidPacketIDAck = 1005 // Send: Invalid packet ID received in ACK message. TerminateReasonInvalidPacketIDNak = 1006 // Send: Invalid packet ID received in NAK message. TerminateReasonCorruptPacketNak = 1007 // Send: Invalid NAK packet received. - TerminateReasonExpireTimer = 1008 // Send: EXP timer expired. ) // DialUDT establishes an outbound UDT connection using the existing provided packet connection. It creates a UDT client. diff --git a/udt/udtsocket_cc.go b/udt/udtsocket_cc.go index 82f3ed0..3b050f2 100644 --- a/udt/udtsocket_cc.go +++ b/udt/udtsocket_cc.go @@ -212,8 +212,3 @@ func (s *udtSocketCc) GetMSS() uint { func (s *udtSocketCc) SetACKPeriod(ack time.Duration) { s.socket.recv.ackPeriod.set(ack) } - -// SetRTOPeriod overrides the default EXP timeout calculations waiting for data from the peer -func (s *udtSocketCc) SetRTOPeriod(rto time.Duration) { - s.socket.send.rtoPeriod.set(rto) -} diff --git a/udt/udtsocket_recv.go b/udt/udtsocket_recv.go index a7173f8..7665f14 100644 --- a/udt/udtsocket_recv.go +++ b/udt/udtsocket_recv.go @@ -90,12 +90,6 @@ ACK is used to trigger an acknowledgement (ACK). Its period is set by NAK is used to trigger a negative acknowledgement (NAK). Its period is dynamically updated to 4 * RTT_+ RTTVar + SYN, where RTTVar is the variance of RTT samples. - - EXP is used to trigger data packets retransmission and maintain - connection status. Its period is dynamically updated to N * (4 * RTT - + RTTVar + SYN), where N is the number of continuous timeouts. To - avoid unnecessary timeout, a minimum threshold (e.g., 0.5 second) - should be used in the implementation. */ // ingestAck2 is called to process an ACK2 packet diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index 66a7a15..c662262 100644 --- a/udt/udtsocket_send.go +++ b/udt/udtsocket_send.go @@ -34,24 +34,20 @@ type udtSocketSend struct { sendPktSeq packet.PacketID // the current packet sequence number msgPartialSend *sendMessage // when a message can only partially fit in a socket, this is the remainder msgSeq uint32 // the current message sequence number - expCount uint // number of continuous EXP timeouts. lastRecvTime time.Time // the last time we've heard something from the remote system recvAckSeq packet.PacketID // largest packetID we've received an ACK from sendLossList *receiveLossHeap // loss list sndPeriod atomicDuration // (set by congestion control) delay between sending packets - rtoPeriod atomicDuration // (set by congestion control) override of EXP timer calculations congestWindow atomicUint32 // (set by congestion control) size of the current congestion window (in packets) flowWindowSize uint // negotiated maximum number of unacknowledged packets (in packets) // timers - sndEvent <-chan time.Time // if a packet is recently sent, this timer fires when SND completes - expTimerEvent <-chan time.Time // Fires when we haven't heard from the peer in a while + sndEvent <-chan time.Time // if a packet is recently sent, this timer fires when SND completes } func newUdtSocketSend(s *udtSocket) *udtSocketSend { ss := &udtSocketSend{ socket: s, - expCount: 1, sendPktSeq: s.initPktSeq, sockClosed: s.sockClosed, sendEvent: s.sendEvent, @@ -63,7 +59,6 @@ func newUdtSocketSend(s *udtSocket) *udtSocketSend { sendPktPend: createPacketHeap(), sendLossList: createPacketIDHeap(), } - ss.resetEXP(s.created) go ss.goSendEvent() return ss } @@ -125,8 +120,6 @@ func (s *udtSocketSend) goSendEvent() { if !ok { return } - s.expCount = 1 - s.resetEXP(evt.now) switch sp := evt.pkt.(type) { case *packet.AckPacket: s.ingestAck(sp, evt.now) @@ -138,8 +131,6 @@ func (s *udtSocketSend) goSendEvent() { s.sendState = s.reevalSendState() case _, _ = <-sockClosed: return - case now := <-s.expTimerEvent: // EXP event - s.expEvent(now) case <-s.sndEvent: // SND event s.sndEvent = nil if s.sendState == sendStateSending { @@ -431,52 +422,3 @@ func (s *udtSocketSend) ingestCongestion(p *packet.CongestionPacket, now time.Ti s.sndPeriod.set(s.sndPeriod.get() * 1125 / 1000) //m_iLastDecSeq = s.sendPktSeq } - -func (s *udtSocketSend) resetEXP(now time.Time) { - s.lastRecvTime = now - - var nextExpDurn time.Duration - rtoPeriod := s.rtoPeriod.get() - if rtoPeriod > 0 { - nextExpDurn = rtoPeriod - } else { - rtt, rttVar := s.socket.getRTT() - nextExpDurn = (time.Duration(s.expCount*(rtt+4*rttVar))*time.Microsecond + s.socket.Config.SynTime) - minExpTime := time.Duration(s.expCount) * minEXPinterval - if nextExpDurn < minExpTime { - nextExpDurn = minExpTime - } - } - s.expTimerEvent = time.After(nextExpDurn) -} - -// we've just had the EXP timer expire, see what we can do to recover this -func (s *udtSocketSend) expEvent(currTime time.Time) { - - // Haven't receive any information from the peer, is it dead?! - // timeout: at least 16 expirations and must be greater than 10 seconds - if (s.expCount > 16) && (currTime.Sub(s.lastRecvTime) > 5*time.Second) { - // Connection is broken. - s.shutdownEvent <- shutdownMessage{sockState: sockStateTimeout, permitLinger: true, reason: TerminateReasonExpireTimer} - return - } - - // sender: Insert all the packets sent after last received acknowledgement into the sender loss list. - // recver: Send out a keep-alive packet - if s.sendPktPend.Count() > 0 { - if s.sendLossList.Count() == 0 { - // resend all unacknowledged packets on timeout, but only if there is no packet in the loss list - for span := s.recvAckSeq.Add(1); span != s.sendPktSeq.Add(1); span.Incr() { - s.sendLossList.Add(recvLossEntry{packetID: packet.PacketID{Seq: span.Seq}}) - } - } - s.socket.cong.onTimeout() - s.sendState = sendStateProcessDrop // immediately restart transmission - } else { - s.sendPacket <- &packet.KeepAlivePacket{} - } - - s.expCount++ - // Reset last response time since we just sent a heart-beat. - s.resetEXP(currTime) -}