From 85a93e9602265df5b96c1e7199f0c10203d268f4 Mon Sep 17 00:00:00 2001 From: Darien Raymond Date: Sun, 3 Dec 2017 14:56:00 +0100 Subject: [PATCH] cleanup kcp --- transport/internet/kcp/congestion.go | 155 --------------------------- transport/internet/kcp/receiving.go | 86 +++++++-------- transport/internet/kcp/sending.go | 128 +++++++++++----------- 3 files changed, 107 insertions(+), 262 deletions(-) delete mode 100644 transport/internet/kcp/congestion.go diff --git a/transport/internet/kcp/congestion.go b/transport/internet/kcp/congestion.go deleted file mode 100644 index 99525fd37..000000000 --- a/transport/internet/kcp/congestion.go +++ /dev/null @@ -1,155 +0,0 @@ -package kcp - -import ( - "sync" -) - -const ( - defaultRTT = 100 - queueSize = 10 -) - -type Queue struct { - value [queueSize]uint32 - start uint32 - length uint32 -} - -func (v *Queue) Push(value uint32) { - if v.length < queueSize { - v.value[v.length] = value - v.length++ - return - } - v.value[v.start] = value - v.start++ - if v.start == queueSize { - v.start = 0 - } -} - -func (v *Queue) Max() uint32 { - max := v.value[0] - for i := 1; i < queueSize; i++ { - if v.value[i] > max { - max = v.value[i] - } - } - return max -} - -func (v *Queue) Min() uint32 { - max := v.value[0] - for i := 1; i < queueSize; i++ { - if v.value[i] < max { - max = v.value[i] - } - } - return max -} - -type CongestionState byte - -const ( - CongestionStateRTTProbe CongestionState = iota - CongestionStateBandwidthProbe - CongestionStateTransfer -) - -type Congestion struct { - sync.RWMutex - - state CongestionState - stateSince uint32 - limit uint32 // bytes per 1000 seconds - - rtt uint32 // millisec - rttHistory Queue - rttUpdateTime uint32 - - initialThroughput uint32 // bytes per 1000 seconds - - cycleStartTime uint32 - cycleBytesConfirmed uint32 - cycleBytesSent uint32 - cycleBytesLimit uint32 - - cycle uint32 - bestCycleBytesConfirmed uint32 - bestCycleBytesSent uint32 -} - -func (v *Congestion) SetState(current uint32, state CongestionState) { - v.state = state - v.stateSince = current -} - -func (v *Congestion) Update(current uint32) { - switch v.state { - case CongestionStateRTTProbe: - if v.rtt > 0 { - v.SetState(current, CongestionStateBandwidthProbe) - v.cycleStartTime = current - v.cycleBytesConfirmed = 0 - v.cycleBytesSent = 0 - v.cycleBytesLimit = v.initialThroughput * v.rtt / 1000 / 1000 - } - case CongestionStateBandwidthProbe: - if current-v.cycleStartTime >= v.rtt { - - } - } -} - -func (v *Congestion) AddBytesConfirmed(current uint32, bytesConfirmed uint32) { - v.Lock() - defer v.Unlock() - - v.cycleBytesConfirmed += bytesConfirmed - v.Update(current) -} - -func (v *Congestion) UpdateRTT(current uint32, rtt uint32) { - v.Lock() - defer v.Unlock() - - if v.state == CongestionStateRTTProbe || rtt < v.rtt { - v.rtt = rtt - v.rttUpdateTime = current - } - - v.Update(current) -} - -func (v *Congestion) GetBytesLimit() uint32 { - v.RLock() - defer v.RUnlock() - - if v.state == CongestionStateRTTProbe { - return v.initialThroughput/1000/(1000/defaultRTT) - v.cycleBytesSent - } - - return v.cycleBytesLimit -} - -func (v *Congestion) RoundTripTime() uint32 { - v.RLock() - defer v.RUnlock() - - if v.state == CongestionStateRTTProbe { - return defaultRTT - } - - return v.rtt -} - -func (v *Congestion) Timeout() uint32 { - v.RLock() - defer v.RUnlock() - - if v.state == CongestionStateRTTProbe { - return defaultRTT * 3 / 2 - } - - return v.rtt * 3 / 2 -} diff --git a/transport/internet/kcp/receiving.go b/transport/internet/kcp/receiving.go index 8c281bb93..1a9353f02 100644 --- a/transport/internet/kcp/receiving.go +++ b/transport/internet/kcp/receiving.go @@ -166,54 +166,54 @@ func NewReceivingWorker(kcp *Connection) *ReceivingWorker { return worker } -func (v *ReceivingWorker) Release() { - v.Lock() - v.leftOver.Release() - v.Unlock() +func (w *ReceivingWorker) Release() { + w.Lock() + w.leftOver.Release() + w.Unlock() } -func (v *ReceivingWorker) ProcessSendingNext(number uint32) { - v.Lock() - defer v.Unlock() +func (w *ReceivingWorker) ProcessSendingNext(number uint32) { + w.Lock() + defer w.Unlock() - v.acklist.Clear(number) + w.acklist.Clear(number) } -func (v *ReceivingWorker) ProcessSegment(seg *DataSegment) { - v.Lock() - defer v.Unlock() +func (w *ReceivingWorker) ProcessSegment(seg *DataSegment) { + w.Lock() + defer w.Unlock() number := seg.Number - idx := number - v.nextNumber - if idx >= v.windowSize { + idx := number - w.nextNumber + if idx >= w.windowSize { return } - v.acklist.Clear(seg.SendingNext) - v.acklist.Add(number, seg.Timestamp) + w.acklist.Clear(seg.SendingNext) + w.acklist.Add(number, seg.Timestamp) - if !v.window.Set(idx, seg) { + if !w.window.Set(idx, seg) { seg.Release() } } -func (v *ReceivingWorker) ReadMultiBuffer() buf.MultiBuffer { - if v.leftOver != nil { - mb := v.leftOver - v.leftOver = nil +func (w *ReceivingWorker) ReadMultiBuffer() buf.MultiBuffer { + if w.leftOver != nil { + mb := w.leftOver + w.leftOver = nil return mb } mb := buf.NewMultiBufferCap(32) - v.Lock() - defer v.Unlock() + w.Lock() + defer w.Unlock() for { - seg := v.window.RemoveFirst() + seg := w.window.RemoveFirst() if seg == nil { break } - v.window.Advance() - v.nextNumber++ + w.window.Advance() + w.nextNumber++ mb.Append(seg.Data) seg.Data = nil seg.Release() @@ -222,11 +222,11 @@ func (v *ReceivingWorker) ReadMultiBuffer() buf.MultiBuffer { return mb } -func (v *ReceivingWorker) Read(b []byte) int { - mb := v.ReadMultiBuffer() +func (w *ReceivingWorker) Read(b []byte) int { + mb := w.ReadMultiBuffer() nBytes, _ := mb.Read(b) if !mb.IsEmpty() { - v.leftOver = mb + w.leftOver = mb } return nBytes } @@ -244,30 +244,30 @@ func (w *ReceivingWorker) NextNumber() uint32 { return w.nextNumber } -func (v *ReceivingWorker) Flush(current uint32) { - v.Lock() - defer v.Unlock() +func (w *ReceivingWorker) Flush(current uint32) { + w.Lock() + defer w.Unlock() - v.acklist.Flush(current, v.conn.roundTrip.Timeout()) + w.acklist.Flush(current, w.conn.roundTrip.Timeout()) } -func (v *ReceivingWorker) Write(seg Segment) error { +func (w *ReceivingWorker) Write(seg Segment) error { ackSeg := seg.(*AckSegment) - ackSeg.Conv = v.conn.conv - ackSeg.ReceivingNext = v.nextNumber - ackSeg.ReceivingWindow = v.nextNumber + v.windowSize - if v.conn.State() == StateReadyToClose { + ackSeg.Conv = w.conn.conv + ackSeg.ReceivingNext = w.nextNumber + ackSeg.ReceivingWindow = w.nextNumber + w.windowSize + if w.conn.State() == StateReadyToClose { ackSeg.Option = SegmentOptionClose } - return v.conn.output.Write(ackSeg) + return w.conn.output.Write(ackSeg) } -func (v *ReceivingWorker) CloseRead() { +func (*ReceivingWorker) CloseRead() { } -func (v *ReceivingWorker) UpdateNecessary() bool { - v.RLock() - defer v.RUnlock() +func (w *ReceivingWorker) UpdateNecessary() bool { + w.RLock() + defer w.RUnlock() - return len(v.acklist.numbers) > 0 + return len(w.acklist.numbers) > 0 } diff --git a/transport/internet/kcp/sending.go b/transport/internet/kcp/sending.go index 6a022e26d..822a3881e 100644 --- a/transport/internet/kcp/sending.go +++ b/transport/internet/kcp/sending.go @@ -36,94 +36,94 @@ func NewSendingWindow(size uint32, writer SegmentWriter, onPacketLoss func(uint3 return window } -func (v *SendingWindow) Release() { - if v == nil { +func (sw *SendingWindow) Release() { + if sw == nil { return } - v.len = 0 - for _, seg := range v.data { + sw.len = 0 + for _, seg := range sw.data { seg.Release() } } -func (v *SendingWindow) Len() int { - return int(v.len) +func (sw *SendingWindow) Len() int { + return int(sw.len) } -func (v *SendingWindow) IsEmpty() bool { - return v.len == 0 +func (sw *SendingWindow) IsEmpty() bool { + return sw.len == 0 } -func (v *SendingWindow) Size() uint32 { - return v.cap +func (sw *SendingWindow) Size() uint32 { + return sw.cap } -func (v *SendingWindow) IsFull() bool { - return v.len == v.cap +func (sw *SendingWindow) IsFull() bool { + return sw.len == sw.cap } -func (v *SendingWindow) Push(number uint32, data []byte) { - pos := (v.start + v.len) % v.cap - v.data[pos].SetData(data) - v.data[pos].Number = number - v.data[pos].timeout = 0 - v.data[pos].transmit = 0 - v.inuse[pos] = true - if v.len > 0 { - v.next[v.last] = pos - v.prev[pos] = v.last +func (sw *SendingWindow) Push(number uint32, data []byte) { + pos := (sw.start + sw.len) % sw.cap + sw.data[pos].SetData(data) + sw.data[pos].Number = number + sw.data[pos].timeout = 0 + sw.data[pos].transmit = 0 + sw.inuse[pos] = true + if sw.len > 0 { + sw.next[sw.last] = pos + sw.prev[pos] = sw.last } - v.last = pos - v.len++ + sw.last = pos + sw.len++ } -func (v *SendingWindow) FirstNumber() uint32 { - return v.data[v.start].Number +func (sw *SendingWindow) FirstNumber() uint32 { + return sw.data[sw.start].Number } -func (v *SendingWindow) Clear(una uint32) { - for !v.IsEmpty() && v.data[v.start].Number < una { - v.Remove(0) +func (sw *SendingWindow) Clear(una uint32) { + for !sw.IsEmpty() && sw.data[sw.start].Number < una { + sw.Remove(0) } } -func (v *SendingWindow) Remove(idx uint32) bool { - if v.IsEmpty() { +func (sw *SendingWindow) Remove(idx uint32) bool { + if sw.IsEmpty() { return false } - pos := (v.start + idx) % v.cap - if !v.inuse[pos] { + pos := (sw.start + idx) % sw.cap + if !sw.inuse[pos] { return false } - v.inuse[pos] = false - v.totalInFlightSize-- - if pos == v.start && pos == v.last { - v.len = 0 - v.start = 0 - v.last = 0 - } else if pos == v.start { - delta := v.next[pos] - v.start - if v.next[pos] < v.start { - delta = v.next[pos] + v.cap - v.start + sw.inuse[pos] = false + sw.totalInFlightSize-- + if pos == sw.start && pos == sw.last { + sw.len = 0 + sw.start = 0 + sw.last = 0 + } else if pos == sw.start { + delta := sw.next[pos] - sw.start + if sw.next[pos] < sw.start { + delta = sw.next[pos] + sw.cap - sw.start } - v.start = v.next[pos] - v.len -= delta - } else if pos == v.last { - v.last = v.prev[pos] + sw.start = sw.next[pos] + sw.len -= delta + } else if pos == sw.last { + sw.last = sw.prev[pos] } else { - v.next[v.prev[pos]] = v.next[pos] - v.prev[v.next[pos]] = v.prev[pos] + sw.next[sw.prev[pos]] = sw.next[pos] + sw.prev[sw.next[pos]] = sw.prev[pos] } return true } -func (v *SendingWindow) HandleFastAck(number uint32, rto uint32) { - if v.IsEmpty() { +func (sw *SendingWindow) HandleFastAck(number uint32, rto uint32) { + if sw.IsEmpty() { return } - v.Visit(func(seg *DataSegment) bool { + sw.Visit(func(seg *DataSegment) bool { if number == seg.Number || number-seg.Number > 0x7FFFFFFF { return false } @@ -135,33 +135,33 @@ func (v *SendingWindow) HandleFastAck(number uint32, rto uint32) { }) } -func (v *SendingWindow) Visit(visitor func(seg *DataSegment) bool) { - if v.IsEmpty() { +func (sw *SendingWindow) Visit(visitor func(seg *DataSegment) bool) { + if sw.IsEmpty() { return } - for i := v.start; ; i = v.next[i] { - if !visitor(&v.data[i]) || i == v.last { + for i := sw.start; ; i = sw.next[i] { + if !visitor(&sw.data[i]) || i == sw.last { break } } } -func (v *SendingWindow) Flush(current uint32, rto uint32, maxInFlightSize uint32) { - if v.IsEmpty() { +func (sw *SendingWindow) Flush(current uint32, rto uint32, maxInFlightSize uint32) { + if sw.IsEmpty() { return } var lost uint32 var inFlightSize uint32 - v.Visit(func(segment *DataSegment) bool { + sw.Visit(func(segment *DataSegment) bool { if current-segment.timeout >= 0x7FFFFFFF { return true } if segment.transmit == 0 { // First time - v.totalInFlightSize++ + sw.totalInFlightSize++ } else { lost++ } @@ -169,7 +169,7 @@ func (v *SendingWindow) Flush(current uint32, rto uint32, maxInFlightSize uint32 segment.Timestamp = current segment.transmit++ - v.writer.Write(segment) + sw.writer.Write(segment) inFlightSize++ if inFlightSize >= maxInFlightSize { return false @@ -177,9 +177,9 @@ func (v *SendingWindow) Flush(current uint32, rto uint32, maxInFlightSize uint32 return true }) - if v.onPacketLoss != nil && inFlightSize > 0 && v.totalInFlightSize != 0 { - rate := lost * 100 / v.totalInFlightSize - v.onPacketLoss(rate) + if sw.onPacketLoss != nil && inFlightSize > 0 && sw.totalInFlightSize != 0 { + rate := lost * 100 / sw.totalInFlightSize + sw.onPacketLoss(rate) } }