From 7adadf9ad52cfce58079e8e1ef053cf699459390 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Mon, 13 Dec 2021 16:21:53 +0100 Subject: [PATCH] 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 | 15 +++++++-- udt/udtsocket_recv.go | 43 ++++++++++++++++-------- udt/udtsocket_send.go | 73 +++++++++++++++++++++++++++-------------- 3 files changed, 90 insertions(+), 41 deletions(-) diff --git a/udt/ack_history_heap.go b/udt/ack_history_heap.go index ab7d0c4..4d1853e 100644 --- a/udt/ack_history_heap.go +++ b/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] } } diff --git a/udt/udtsocket_recv.go b/udt/udtsocket_recv.go index be87ba5..f8d4160 100644 --- a/udt/udtsocket_recv.go +++ b/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 } diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index 2325cb3..5e41b37 100644 --- a/udt/udtsocket_send.go +++ b/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) } }