diff --git a/transport/internet/kcp/receiving.go b/transport/internet/kcp/receiving.go index 2469063f1..3ff3b1285 100644 --- a/transport/internet/kcp/receiving.go +++ b/transport/internet/kcp/receiving.go @@ -80,14 +80,15 @@ func (this *AckList) Add(number uint32, timestamp uint32) { func (this *AckList) Clear(una uint32) { count := 0 for i := 0; i < len(this.numbers); i++ { - if this.numbers[i] >= una { - if i != count { - this.numbers[count] = this.numbers[i] - this.timestamps[count] = this.timestamps[i] - this.nextFlush[count] = this.nextFlush[i] - } - count++ + if this.numbers[i] < una { + continue } + if i != count { + this.numbers[count] = this.numbers[i] + this.timestamps[count] = this.timestamps[i] + this.nextFlush[count] = this.nextFlush[i] + } + count++ } if count < len(this.numbers) { this.numbers = this.numbers[:count] @@ -99,15 +100,16 @@ func (this *AckList) Clear(una uint32) { func (this *AckList) Flush(current uint32, rto uint32) { seg := NewAckSegment() for i := 0; i < len(this.numbers) && !seg.IsFull(); i++ { - if this.nextFlush[i] <= current { - seg.PutNumber(this.numbers[i]) - seg.PutTimestamp(this.timestamps[i]) - timeout := rto / 4 - if timeout < 20 { - timeout = 20 - } - this.nextFlush[i] = current + timeout + if this.nextFlush[i] > current { + continue } + seg.PutNumber(this.numbers[i]) + seg.PutTimestamp(this.timestamps[i]) + timeout := rto / 4 + if timeout < 20 { + timeout = 20 + } + this.nextFlush[i] = current + timeout } if seg.Count > 0 { this.writer.Write(seg) diff --git a/transport/internet/kcp/sending.go b/transport/internet/kcp/sending.go index 18036bea2..869d315e9 100644 --- a/transport/internet/kcp/sending.go +++ b/transport/internet/kcp/sending.go @@ -109,21 +109,25 @@ func (this *SendingWindow) Remove(idx uint32) bool { } func (this *SendingWindow) HandleFastAck(number uint32, rto uint32) { - if this.len == 0 { + if this.IsEmpty() { return } + this.Visit(func(seg *DataSegment) bool { + if number == seg.Number || number-seg.Number > 0x7FFFFFFF { + return false + } + + if seg.transmit > 0 && seg.timeout > rto/3 { + seg.timeout -= rto / 3 + } + return true + }) +} + +func (this *SendingWindow) Visit(visitor func(seg *DataSegment) bool) { for i := this.start; ; i = this.next[i] { - seg := &this.data[i] - if number-seg.Number > 0x7FFFFFFF { - break - } - if number != seg.Number { - if seg.transmit > 0 && seg.timeout > rto/3 { - seg.timeout -= rto / 3 - } - } - if i == this.last { + if !visitor(&this.data[i]) || i == this.last { break } } @@ -137,33 +141,27 @@ func (this *SendingWindow) Flush(current uint32, rto uint32, maxInFlightSize uin var lost uint32 var inFlightSize uint32 - for i := this.start; ; i = this.next[i] { - segment := &this.data[i] - needsend := false - if current-segment.timeout < 0x7FFFFFFF { - if segment.transmit == 0 { - // First time - this.totalInFlightSize++ - } else { - lost++ - } - needsend = true - segment.timeout = current + rto + this.Visit(func(segment *DataSegment) bool { + if current-segment.timeout >= 0x7FFFFFFF { + return true } + if segment.transmit == 0 { + // First time + this.totalInFlightSize++ + } else { + lost++ + } + segment.timeout = current + rto - if needsend { - segment.Timestamp = current - segment.transmit++ - this.writer.Write(segment) - inFlightSize++ - if inFlightSize >= maxInFlightSize { - break - } + segment.Timestamp = current + segment.transmit++ + this.writer.Write(segment) + inFlightSize++ + if inFlightSize >= maxInFlightSize { + return false } - if i == this.last { - break - } - } + return true + }) if this.onPacketLoss != nil && inFlightSize > 0 && this.totalInFlightSize != 0 { rate := lost * 100 / this.totalInFlightSize