From 94ed38348993d04e7c525ca888627f6b59d04a68 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Sun, 20 Feb 2022 18:24:42 +0100 Subject: [PATCH] UDT: Fix critical bug in NativeCongestionControl not using pointer. ackEvent immediate parameter. --- udt/congestion_native.go | 30 ++++++++++++++---------------- udt/udtsocket_recv.go | 24 ++++++++++++++---------- 2 files changed, 28 insertions(+), 26 deletions(-) diff --git a/udt/congestion_native.go b/udt/congestion_native.go index 450925b..44847aa 100644 --- a/udt/congestion_native.go +++ b/udt/congestion_native.go @@ -24,7 +24,7 @@ type NativeCongestionControl struct { } // Init to be called (only) at the start of a UDT connection. -func (ncc NativeCongestionControl) Init(parms CongestionControlParms, synTime time.Duration) { +func (ncc *NativeCongestionControl) Init(parms CongestionControlParms, synTime time.Duration) { ncc.rcInterval = synTime ncc.lastRCTime = time.Now() parms.SetACKPeriod(ncc.rcInterval) @@ -32,7 +32,7 @@ func (ncc NativeCongestionControl) Init(parms CongestionControlParms, synTime ti // This value should be adjusted at runtime according to congestion. parms.SetACKInterval(4) - ncc.slowStart = true + ncc.slowStart = false ncc.lastAck = parms.GetSndCurrSeqNo() ncc.loss = false ncc.lastDecSeq = ncc.lastAck.Add(-1) @@ -41,17 +41,17 @@ func (ncc NativeCongestionControl) Init(parms CongestionControlParms, synTime ti ncc.nakCount = 0 ncc.decRandom = 1 - parms.SetCongestionWindowSize(16) + parms.SetCongestionWindowSize(32) parms.SetPacketSendPeriod(1 * time.Microsecond) } // Close to be called when a UDT connection is closed. -func (ncc NativeCongestionControl) Close(parms CongestionControlParms) { +func (ncc *NativeCongestionControl) Close(parms CongestionControlParms) { // nothing done for this event } // OnACK to be called when an ACK packet is received -func (ncc NativeCongestionControl) OnACK(parms CongestionControlParms, ack packet.PacketID) { +func (ncc *NativeCongestionControl) OnACK(parms CongestionControlParms, ack packet.PacketID) { currTime := time.Now() if currTime.Sub(ncc.lastRCTime) < ncc.rcInterval { return @@ -138,7 +138,7 @@ func (ncc NativeCongestionControl) OnACK(parms CongestionControlParms, ack packe } // OnNAK to be called when a loss report is received -func (ncc NativeCongestionControl) OnNAK(parms CongestionControlParms, losslist []packet.PacketID) { +func (ncc *NativeCongestionControl) OnNAK(parms CongestionControlParms, losslist []packet.PacketID) { // If it is in slow start phase, set inter-packet interval to 1/recvrate. Slow start ends. Stop. if ncc.slowStart { ncc.slowStart = false @@ -201,7 +201,7 @@ func (ncc NativeCongestionControl) OnNAK(parms CongestionControlParms, losslist } // OnTimeout to be called when a timeout event occurs -func (ncc NativeCongestionControl) OnTimeout(parms CongestionControlParms) { +func (ncc *NativeCongestionControl) OnTimeout(parms CongestionControlParms) { if ncc.slowStart { ncc.slowStart = false recvRate, _ := parms.GetReceiveRates() @@ -211,26 +211,24 @@ func (ncc NativeCongestionControl) OnTimeout(parms CongestionControlParms) { parms.SetPacketSendPeriod(time.Duration(float64(time.Microsecond) * float64(parms.GetCongestionWindowSize()) / float64(parms.GetRTT()+ncc.rcInterval))) } } else { - /* - pktSendPeriod := parms.GetPacketSendPeriod() - ncc.lastDecPeriod = pktSendPeriod - parms.SetPacketSendPeriod(math.Ceil(pktSendPeriod * 2)) - ncc.lastDecSeq = ncc.lastAck - */ + pktSendPeriod := parms.GetPacketSendPeriod() + ncc.lastDecPeriod = pktSendPeriod + parms.SetPacketSendPeriod(time.Duration(pktSendPeriod * 2)) + ncc.lastDecSeq = ncc.lastAck } } // OnPktSent to be called when data is sent -func (ncc NativeCongestionControl) OnPktSent(parms CongestionControlParms, pkt packet.Packet) { +func (ncc *NativeCongestionControl) OnPktSent(parms CongestionControlParms, pkt packet.Packet) { // nothing done for this event } // OnPktRecv to be called when a data is received -func (ncc NativeCongestionControl) OnPktRecv(parms CongestionControlParms, pkt packet.DataPacket) { +func (ncc *NativeCongestionControl) OnPktRecv(parms CongestionControlParms, pkt packet.DataPacket) { // nothing done for this event } // OnCustomMsg to process a user-defined packet -func (ncc NativeCongestionControl) OnCustomMsg(parms CongestionControlParms, pkt packet.UserDefControlPacket) { +func (ncc *NativeCongestionControl) OnCustomMsg(parms CongestionControlParms, pkt packet.UserDefControlPacket) { // nothing done for this event } diff --git a/udt/udtsocket_recv.go b/udt/udtsocket_recv.go index 64661cb..d110b18 100644 --- a/udt/udtsocket_recv.go +++ b/udt/udtsocket_recv.go @@ -145,7 +145,7 @@ func (s *udtSocketRecv) ingestMsgDropReq(p *packet.MsgDropReqPacket, now time.Ti // try to push any pending packets out, now that we have dropped any blocking packets for _, nextPkt := range s.recvPktPend.Range(stopSeq, s.nextSequenceExpect) { - if !s.attemptProcessPacket(nextPkt.pkt, false) { + if !s.attemptProcessPacket(nextPkt.pkt, false, false) { break } } @@ -184,6 +184,7 @@ func (s *udtSocketRecv) ingestData(p *packet.DataPacket, now time.Time) { } } s.recvLastArrival = now + var ackImmediate bool // If the incoming sequence number is greater than the expected one, treat all sequence numbers in the middle as lost (add to lost list) and send a NAK. seqDiff := p.Seq.BlindDiff(s.nextSequenceExpect) @@ -201,19 +202,20 @@ func (s *udtSocketRecv) ingestData(p *packet.DataPacket, now time.Time) { if !s.recvLossList.Remove(p.Seq.Seq) { return // already previously received packet -- ignore } + ackImmediate = true } else { s.nextSequenceExpect = p.Seq.Add(1) } if s.socket.isDatagram && p.Seq == s.lastSequence.Add(1) { s.lastSequence = p.Seq - s.ackEvent() // Need special sending for datagram, otherwise below code would only send it out after all pieces are received. + s.ackEvent(false) // Need special sending for datagram, otherwise below code would only send it out after all pieces are received. } - s.attemptProcessPacket(p, true) + s.attemptProcessPacket(p, true, ackImmediate) } -func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) bool { +func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew, ackImmediate bool) bool { var pieces []*packet.DataPacket var success bool @@ -239,7 +241,7 @@ func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) b } s.lastSequence = pieces[len(pieces)-1].Seq - s.ackEvent() + s.ackEvent(ackImmediate) // reassemble the data by appending it from all the pieces var msg []byte @@ -465,7 +467,7 @@ func (s *udtSocketRecv) ingestError(p *packet.ErrPacket) { } // ackEvent sends an ACK message if appropriate. It informs the remote peer about the last packet received without loss. -func (s *udtSocketRecv) ackEvent() { +func (s *udtSocketRecv) ackEvent(immediate bool) { s.unackPktCount++ // The ack number is excluding. @@ -477,10 +479,12 @@ func (s *udtSocketRecv) ackEvent() { } // Check if the threshold to send is reached, if used. Note that sendACK is called revery SynTime. - ackInterval := uint(s.ackInterval.get()) - if (ackInterval > 0) && (ackInterval > s.unackPktCount) { - s.sentAck = ack // This is needed for resendACKTimer to pick it up in case no ackInterval count of packets are immediately sent. - return + if !immediate { + ackInterval := uint(s.ackInterval.get()) + if (ackInterval > 0) && (ackInterval > s.unackPktCount) { + s.sentAck = ack // This is needed for resendACKTimer to pick it up in case no ackInterval count of packets are immediately sent. + return + } } s.sendACK(ack)