UDT: Fix critical bugs that appear during high packet loss. Recv: Implement resending ACKs (until ACK2'ed). Send: Implement unsolicited data resending of unacked data packets. Double sending time each try to prevent ddos.

udt/ack_history_heap.go

@@ -38,13 +38,22 @@ func (heap *ackHistoryHeap) Remove(sequence uint32) (found *ackHistoryEntry) {
 	heap.Lock()
 	defer heap.Unlock()
 
+	for n := range heap.list {
+		if heap.list[n].ackID == sequence {
+			found = &heap.list[n]
+		}
+	}
+
+	if found == nil {
+		return nil
+	}
+
+	// if found, automatically remove all entries with a lower lastPacket
 	var newList []ackHistoryEntry
 
 	for n := range heap.list {
-		if heap.list[n].ackID != sequence {
+		if heap.list[n].ackID != sequence && heap.list[n].lastPacket.IsBigger(found.lastPacket) {
 			newList = append(newList, heap.list[n])
-		} else {
-			found = &heap.list[n]
 		}
 	}
 

udt/udtsocket_recv.go

@@ -16,8 +16,7 @@ type udtSocketRecv struct {
 
 	nextSequenceExpect packet.PacketID  // the peer's next largest packet ID expected.
 	lastSequence       packet.PacketID  // the peer's last received packet ID before any loss events
-	lastACK            uint32           // last ACK packet we've sent
-	largestACK         uint32           // largest ACK packet we've sent that has been acknowledged (by an ACK2).
+	lastACKID          uint32           // last ACK packet we've sent
 	recvPktPend        *sendPacketHeap  // list of packets that are waiting to be processed.
 	recvLossList       *receiveLossHeap // loss list.
 	ackHistory         *ackHistoryHeap  // list of sent ACKs.
@@ -32,6 +31,8 @@ type udtSocketRecv struct {
 	ackLinkInfoSent    time.Time        // when link info was sent in ACK packet last time
 	resendNAKTimer     <-chan time.Time // Timer for resending outgoing NAK
 	resendNAKTicker    time.Ticker      // Ticker for resending outgoing NAK
+	resendACKTimer     <-chan time.Time // Timer for resending outgoing ACK
+	resendACKTicker    time.Ticker      // Ticker for resending outgoing ACK
 }
 
 func newUdtSocketRecv(s *udtSocket) *udtSocketRecv {
@@ -45,6 +46,7 @@ func newUdtSocketRecv(s *udtSocket) *udtSocketRecv {
 		recvLossList:   createPacketIDHeap(),
 		ackHistory:     createHistoryHeap(),
 		resendNAKTimer: make(chan time.Time),
+		resendACKTimer: make(chan time.Time),
 	}
 	go sr.goReceiveEvent()
 	return sr
@@ -76,8 +78,9 @@ func (s *udtSocketRecv) goReceiveEvent() {
 			case *packet.ErrPacket:
 				s.ingestError(sp)
 			}
-		case _, _ = <-sockClosed: // socket is closed, leave now
+		case <-sockClosed: // socket is closed, leave now
 			s.resendNAKTicker.Stop()
+			s.resendACKTicker.Stop()
 			return
 		case <-s.resendNAKTimer:
 			if first, valid := s.recvLossList.FirstSequence(); valid {
@@ -87,6 +90,14 @@ func (s *udtSocketRecv) goReceiveEvent() {
 				s.resendNAKTimer = make(chan time.Time)
 				s.resendNAKTicker.Stop()
 			}
+		case <-s.resendACKTimer:
+			if s.recvAck2.IsLess(s.sentAck) {
+				s.sendACK(s.sentAck)
+			} else {
+				// the trigger should not be activated if there is no unacknowledged ACK list, but in case it happens, deactivate it
+				s.resendACKTimer = make(chan time.Time)
+				s.resendACKTicker.Stop()
+			}
 		}
 	}
 }
@@ -106,21 +117,21 @@ ACK is used to trigger an acknowledgement (ACK). Its period is set by
 
 // ingestAck2 is called to process an ACK2 packet
 func (s *udtSocketRecv) ingestAck2(p *packet.Ack2Packet, now time.Time) {
-	ackHistEntry := s.ackHistory.Remove(p.AckSeqNo)
+	ackHistEntry := s.ackHistory.Remove(p.AckSeqNo) // this also removes all other unacknoweldged ACKs with a lower lastPacket
 	if ackHistEntry == nil {
 		return // this ACK not found
 	}
-	if s.recvAck2.BlindDiff(ackHistEntry.lastPacket) < 0 {
-		s.recvAck2 = ackHistEntry.lastPacket
-	}
 
-	// Update the largest ACK number ever been acknowledged.
-	if s.largestACK < p.AckSeqNo {
-		s.largestACK = p.AckSeqNo
-	}
+	s.recvAck2 = ackHistEntry.lastPacket
 
 	s.socket.applyRTT(uint(now.Sub(ackHistEntry.sendTime) / time.Microsecond))
 
+	if s.ackHistory.Count() == 0 {
+		// deactivate ACK timer, nothing expected for now
+		s.resendACKTimer = make(chan time.Time)
+		s.resendACKTicker.Stop()
+	}
+
 	//s.rto = 4 * s.rtt + s.rttVar
 }
 
@@ -426,9 +437,9 @@ func (s *udtSocketRecv) getRcvSpeeds() (recvSpeed, bandwidth int) {
 func (s *udtSocketRecv) sendACK(ack packet.PacketID) {
 	s.sentAck = ack
 
-	s.lastACK++
+	s.lastACKID++
 	s.ackHistory.Add(ackHistoryEntry{
-		ackID:      s.lastACK,
+		ackID:      s.lastACKID,
 		lastPacket: ack,
 		sendTime:   time.Now(),
 	})
@@ -442,7 +453,7 @@ func (s *udtSocketRecv) sendACK(ack packet.PacketID) {
 	}
 
 	p := &packet.AckPacket{
-		AckSeqNo:  s.lastACK,
+		AckSeqNo:  s.lastACKID,
 		PktSeqHi:  ack,
 		Rtt:       uint32(rtt),
 		RttVar:    uint32(rttVar),
@@ -495,4 +506,8 @@ func (s *udtSocketRecv) ackEvent() {
 
 	s.sendACK(ack)
 	s.unackPktCount = 0
+
+	// set the timer for constantly resending ACKs for the highest sequence ID
+	s.resendACKTicker = *time.NewTicker(s.socket.Config.SynTime)
+	s.resendACKTimer = s.resendACKTicker.C
 }

udt/udtsocket_send.go

@@ -29,33 +29,35 @@ type udtSocketSend struct {
 	shutdownEvent chan<- shutdownMessage // channel signals the connection to be shutdown
 	socket        *udtSocket
 
-	sendState      sendState        // current sender state
-	sendPktPend    *sendPacketHeap  // list of packets that have been sent but not yet acknowledged
-	sendPktSeq     packet.PacketID  // the current packet sequence number
-	msgRemainder   *sendMessage     // when a message can only partially fit in a socket, this is the remainder
-	msgSeq         uint32           // the current message sequence number
-	lastSendTime   time.Time        // the last time we've sent a data packet to the remote system
-	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. New entries added via incoming NAK.
-	sndPeriod      atomicDuration   // (set by congestion control) delay between sending packets
-	congestWindow  atomicUint32     // (set by congestion control) size of the current congestion window (in packets)
-	flowWindowSize uint             // negotiated maximum number of unacknowledged packets (in packets)
+	sendState       sendState        // current sender state
+	sendPktPend     *sendPacketHeap  // list of packets that have been sent but not yet acknowledged
+	sendPktSeq      packet.PacketID  // the current packet sequence number
+	msgRemainder    *sendMessage     // when a message can only partially fit in a socket, this is the remainder
+	msgSeq          uint32           // the current message sequence number
+	lastSendTime    time.Time        // the last time we've sent a data packet to the remote system
+	recvAckSeq      packet.PacketID  // largest packetID we've received an ACK from
+	sendLossList    *receiveLossHeap // loss list. New entries added via incoming NAK.
+	sndPeriod       atomicDuration   // (set by congestion control) delay between sending packets
+	congestWindow   atomicUint32     // (set by congestion control) size of the current congestion window (in packets)
+	flowWindowSize  uint             // negotiated maximum number of unacknowledged packets (in packets)
+	resendDataTimer <-chan time.Time // Timer for resending outgoing data packets
+	resendDataTime  time.Duration    // Doubles after every send to prevent ddos
 }
 
 func newUdtSocketSend(s *udtSocket) *udtSocketSend {
 	ss := &udtSocketSend{
-		socket:         s,
-		sendPktSeq:     s.initPktSeq,
-		sockClosed:     s.sockClosed,
-		sendEvent:      s.sendEvent,
-		messageOut:     s.messageOut,
-		congestWindow:  atomicUint32{val: 16},
-		flowWindowSize: s.maxFlowWinSize,
-		sendPacket:     s.sendPacket,
-		shutdownEvent:  s.shutdownEvent,
-		sendPktPend:    createPacketHeap(),
-		sendLossList:   createPacketIDHeap(),
+		socket:          s,
+		sendPktSeq:      s.initPktSeq,
+		sockClosed:      s.sockClosed,
+		sendEvent:       s.sendEvent,
+		messageOut:      s.messageOut,
+		congestWindow:   atomicUint32{val: 16},
+		flowWindowSize:  s.maxFlowWinSize,
+		sendPacket:      s.sendPacket,
+		shutdownEvent:   s.shutdownEvent,
+		sendPktPend:     createPacketHeap(),
+		sendLossList:    createPacketIDHeap(),
+		resendDataTimer: make(chan time.Time),
 	}
 	go ss.goSendEvent()
 	return ss
@@ -129,7 +131,7 @@ func (s *udtSocketSend) goSendEvent() {
 			}
 
 		case sendStateWaiting:
-			// Destination is full (congested). Do not use event timer, do not check for new messages. Only wait for incoming ACKs.
+			// Destination is full (congested). Do not use event timer, do not check for new messages. Only wait for incoming ACKs + resend data packets.
 
 		case sendStateProcessDrop:
 			// Immediately resend any missing packets. The status will only be updated by incoming ACKs.
@@ -169,6 +171,16 @@ func (s *udtSocketSend) goSendEvent() {
 			s.sendPacket <- &packet.ShutdownPacket{}
 			s.shutdownEvent <- shutdownMessage{sockState: sockStateClosed, permitLinger: !s.socket.isServer, reason: TerminateReasonCannotProcessOutgoing}
 			return
+
+		case <-s.resendDataTimer:
+			// Resend data that was not acknowledged yet.
+			for _, dp := range s.sendPktPend.list {
+				s.sendPacket <- dp.pkt
+			}
+
+			// to prevent ddos, always double the time
+			s.resendDataTime = s.resendDataTime * 2
+			s.resendDataTimer = time.NewTimer(s.resendDataTime).C
 		}
 	}
 }
@@ -182,9 +194,19 @@ func (s *udtSocketSend) reevalSendState() sendState {
 	}
 	if uint(s.sendPktPend.Count()) > cwnd {
 		s.sendState = sendStateWaiting
+
+		// set the timer for constantly resending data packets until ACKed
+		s.resendDataTime = s.socket.Config.SynTime
+		s.resendDataTimer = time.NewTimer(s.resendDataTime).C
+
 		return s.sendState
 	}
 
+	if s.sendState == sendStateWaiting {
+		// constant resending no longer needed
+		s.resendDataTimer = make(chan time.Time)
+	}
+
 	// is the current packet data to send empty? Switch to idle in this case.
 	if s.msgRemainder == nil {
 		s.sendState = sendStateIdle
@@ -460,6 +482,9 @@ func (s *udtSocketSend) ingestNak(p *packet.NakPacket, now time.Time) {
 	// Some loss entries may be discarded if out of date (already ACK received), so make sure loss list contains entries before changing the sending state.
 	if s.sendLossList.Count() > 0 {
 		s.sendState = sendStateProcessDrop // immediately restart transmission
+
+		// resending now orderly handled via NAKs instead of constant data packet resending
+		s.resendDataTimer = make(chan time.Time)
 	}
 }