Peernet-Archive/core
1
0
Fork 0
mirror of https://github.com/PeernetOfficial/core.git synced 2026-07-18 11:17:49 +01:00
Code Issues Packages Projects Releases Wiki Activity

UDT: Fix critical bug in NativeCongestionControl not using pointer. ackEvent immediate parameter.

Browse Source
This commit is contained in:
Kleissner
2022-02-20 18:24:42 +01:00
parent a41de8a4db
commit 94ed383489
2 changed files with 28 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
udt/congestion_native.go
View File

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

24
udt/udtsocket_recv.go
View File

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