UDT: Remove buggy EXP code. It caused wrong termination. It is not needed; the Peernet protocol handles connectivity.

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

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.

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)
-}

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

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)
-}