1
0
mirror of https://github.com/v2fly/v2ray-core.git synced 2024-12-23 02:26:49 -05:00
v2fly/transport/internet/kcp/sending.go

388 lines
7.8 KiB
Go
Raw Normal View History

2016-06-26 17:51:17 -04:00
package kcp
2016-07-03 16:14:38 -04:00
import (
"sync"
)
2016-07-01 05:57:13 -04:00
type SendingWindow struct {
start uint32
cap uint32
len uint32
last uint32
2016-11-01 07:07:20 -04:00
data []DataSegment
inuse []bool
prev []uint32
next []uint32
2016-07-01 05:57:13 -04:00
2016-07-04 09:34:14 -04:00
totalInFlightSize uint32
writer SegmentWriter
onPacketLoss func(uint32)
2016-07-01 05:57:13 -04:00
}
2016-07-04 09:54:18 -04:00
func NewSendingWindow(size uint32, writer SegmentWriter, onPacketLoss func(uint32)) *SendingWindow {
2016-07-01 05:57:13 -04:00
window := &SendingWindow{
2016-07-03 16:14:38 -04:00
start: 0,
cap: size,
len: 0,
last: 0,
2016-11-01 07:07:20 -04:00
data: make([]DataSegment, size),
2016-07-03 16:14:38 -04:00
prev: make([]uint32, size),
next: make([]uint32, size),
2016-11-01 07:07:20 -04:00
inuse: make([]bool, size),
2016-07-03 16:14:38 -04:00
writer: writer,
onPacketLoss: onPacketLoss,
2016-07-01 05:57:13 -04:00
}
return window
}
2016-11-27 15:39:09 -05:00
func (v *SendingWindow) Release() {
if v == nil {
2016-11-21 16:41:12 -05:00
return
}
2016-12-06 18:31:01 -05:00
v.len = 0
2016-11-27 15:39:09 -05:00
for _, seg := range v.data {
2016-11-21 16:41:12 -05:00
seg.Release()
}
}
2016-11-27 15:39:09 -05:00
func (v *SendingWindow) Len() int {
return int(v.len)
2016-07-01 05:57:13 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWindow) IsEmpty() bool {
return v.len == 0
2016-07-12 17:54:54 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWindow) Size() uint32 {
return v.cap
2016-07-04 09:54:18 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWindow) IsFull() bool {
return v.len == v.cap
2016-07-04 09:54:18 -04:00
}
2016-11-27 15:39:09 -05:00
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
2016-07-01 05:57:13 -04:00
}
2016-11-27 15:39:09 -05:00
v.last = pos
v.len++
2016-07-01 05:57:13 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWindow) FirstNumber() uint32 {
return v.data[v.start].Number
2016-07-01 05:57:13 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWindow) Clear(una uint32) {
for !v.IsEmpty() && v.data[v.start].Number < una {
v.Remove(0)
2016-07-01 05:57:13 -04:00
}
}
2016-11-27 15:39:09 -05:00
func (v *SendingWindow) Remove(idx uint32) bool {
2016-12-08 16:49:24 -05:00
if v.IsEmpty() {
2016-11-13 16:27:58 -05:00
return false
2016-07-01 17:27:57 -04:00
}
2016-11-27 15:39:09 -05:00
pos := (v.start + idx) % v.cap
if !v.inuse[pos] {
2016-11-13 16:27:58 -05:00
return false
2016-07-01 06:12:32 -04:00
}
2016-11-27 15:39:09 -05:00
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
2016-07-01 05:57:13 -04:00
}
2016-11-27 15:39:09 -05:00
v.start = v.next[pos]
v.len -= delta
} else if pos == v.last {
v.last = v.prev[pos]
2016-07-01 05:57:13 -04:00
} else {
2016-11-27 15:39:09 -05:00
v.next[v.prev[pos]] = v.next[pos]
v.prev[v.next[pos]] = v.prev[pos]
2016-07-01 05:57:13 -04:00
}
2016-11-13 16:27:58 -05:00
return true
2016-07-01 05:57:13 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWindow) HandleFastAck(number uint32, rto uint32) {
if v.IsEmpty() {
2016-07-01 17:27:57 -04:00
return
}
2016-07-01 06:12:32 -04:00
2016-11-27 15:39:09 -05:00
v.Visit(func(seg *DataSegment) bool {
2016-11-18 10:19:13 -05:00
if number == seg.Number || number-seg.Number > 0x7FFFFFFF {
return false
2016-07-01 05:57:13 -04:00
}
2016-11-18 10:19:13 -05:00
if seg.transmit > 0 && seg.timeout > rto/3 {
seg.timeout -= rto / 3
2016-07-01 05:57:13 -04:00
}
2016-11-18 10:19:13 -05:00
return true
})
}
2016-11-27 15:39:09 -05:00
func (v *SendingWindow) Visit(visitor func(seg *DataSegment) bool) {
2016-12-06 18:31:01 -05:00
if v.IsEmpty() {
return
}
2016-11-27 15:39:09 -05:00
for i := v.start; ; i = v.next[i] {
if !visitor(&v.data[i]) || i == v.last {
2016-07-01 05:57:13 -04:00
break
}
}
}
2016-11-27 15:39:09 -05:00
func (v *SendingWindow) Flush(current uint32, rto uint32, maxInFlightSize uint32) {
if v.IsEmpty() {
2016-07-03 16:14:38 -04:00
return
2016-07-01 06:12:32 -04:00
}
2016-07-04 09:34:14 -04:00
var lost uint32
2016-07-04 07:37:42 -04:00
var inFlightSize uint32
2016-07-01 05:57:13 -04:00
2016-11-27 15:39:09 -05:00
v.Visit(func(segment *DataSegment) bool {
2016-11-18 10:19:13 -05:00
if current-segment.timeout >= 0x7FFFFFFF {
return true
2016-07-01 05:57:13 -04:00
}
2016-11-18 10:19:13 -05:00
if segment.transmit == 0 {
// First time
2016-11-27 15:39:09 -05:00
v.totalInFlightSize++
2016-11-18 10:19:13 -05:00
} else {
lost++
2016-07-01 05:57:13 -04:00
}
2016-11-18 10:19:13 -05:00
segment.timeout = current + rto
segment.Timestamp = current
segment.transmit++
2016-11-27 15:39:09 -05:00
v.writer.Write(segment)
2016-11-18 10:19:13 -05:00
inFlightSize++
if inFlightSize >= maxInFlightSize {
return false
2016-07-01 05:57:13 -04:00
}
2016-11-18 10:19:13 -05:00
return true
})
2016-07-01 05:57:13 -04:00
2016-11-27 15:39:09 -05:00
if v.onPacketLoss != nil && inFlightSize > 0 && v.totalInFlightSize != 0 {
rate := lost * 100 / v.totalInFlightSize
v.onPacketLoss(rate)
2016-07-04 09:34:14 -04:00
}
2016-07-01 05:57:13 -04:00
}
2016-07-03 16:14:38 -04:00
type SendingWorker struct {
2016-07-12 11:56:36 -04:00
sync.RWMutex
2016-10-11 06:24:19 -04:00
conn *Connection
window *SendingWindow
firstUnacknowledged uint32
firstUnacknowledgedUpdated bool
nextNumber uint32
remoteNextNumber uint32
controlWindow uint32
fastResend uint32
2016-07-03 16:14:38 -04:00
}
2016-07-05 17:02:52 -04:00
func NewSendingWorker(kcp *Connection) *SendingWorker {
2016-07-03 16:14:38 -04:00
worker := &SendingWorker{
2016-07-05 17:02:52 -04:00
conn: kcp,
2016-07-03 16:14:38 -04:00
fastResend: 2,
remoteNextNumber: 32,
2016-10-02 17:43:58 -04:00
controlWindow: kcp.Config.GetSendingInFlightSize(),
2016-07-03 16:14:38 -04:00
}
2016-10-02 17:43:58 -04:00
worker.window = NewSendingWindow(kcp.Config.GetSendingBufferSize(), worker, worker.OnPacketLoss)
2016-07-03 16:14:38 -04:00
return worker
}
2016-11-27 15:39:09 -05:00
func (v *SendingWorker) Release() {
2017-02-17 18:04:25 -05:00
v.Lock()
2016-11-27 15:39:09 -05:00
v.window.Release()
2017-02-17 18:04:25 -05:00
v.Unlock()
2016-11-21 16:41:12 -05:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWorker) ProcessReceivingNext(nextNumber uint32) {
v.Lock()
defer v.Unlock()
2016-07-03 16:14:38 -04:00
2016-11-27 15:39:09 -05:00
v.ProcessReceivingNextWithoutLock(nextNumber)
2016-07-06 10:36:15 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWorker) ProcessReceivingNextWithoutLock(nextNumber uint32) {
v.window.Clear(nextNumber)
v.FindFirstUnacknowledged()
2016-07-03 16:14:38 -04:00
}
2016-08-24 05:17:42 -04:00
// Private: Visible for testing.
2016-11-27 15:39:09 -05:00
func (v *SendingWorker) FindFirstUnacknowledged() {
first := v.firstUnacknowledged
if !v.window.IsEmpty() {
v.firstUnacknowledged = v.window.FirstNumber()
2016-07-03 16:14:38 -04:00
} else {
2016-11-27 15:39:09 -05:00
v.firstUnacknowledged = v.nextNumber
2016-07-03 16:14:38 -04:00
}
2016-11-27 15:39:09 -05:00
if first != v.firstUnacknowledged {
v.firstUnacknowledgedUpdated = true
2016-10-11 06:24:19 -04:00
}
2016-07-03 16:14:38 -04:00
}
2017-02-26 09:01:50 -05:00
func (v *SendingWorker) processAck(number uint32) bool {
2016-11-27 15:39:09 -05:00
// number < v.firstUnacknowledged || number >= v.nextNumber
if number-v.firstUnacknowledged > 0x7FFFFFFF || number-v.nextNumber < 0x7FFFFFFF {
2016-11-13 16:27:58 -05:00
return false
2016-07-03 16:14:38 -04:00
}
2016-11-27 15:39:09 -05:00
removed := v.window.Remove(number - v.firstUnacknowledged)
2016-11-13 16:27:58 -05:00
if removed {
2016-11-27 15:39:09 -05:00
v.FindFirstUnacknowledged()
2016-11-13 16:27:58 -05:00
}
return removed
2016-07-03 16:14:38 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWorker) ProcessSegment(current uint32, seg *AckSegment, rto uint32) {
2016-07-15 15:41:15 -04:00
defer seg.Release()
2016-11-27 15:39:09 -05:00
v.Lock()
defer v.Unlock()
2016-07-15 15:41:15 -04:00
2016-11-27 15:39:09 -05:00
if v.remoteNextNumber < seg.ReceivingWindow {
v.remoteNextNumber = seg.ReceivingWindow
2016-07-15 15:41:15 -04:00
}
2016-11-27 15:39:09 -05:00
v.ProcessReceivingNextWithoutLock(seg.ReceivingNext)
2016-07-14 11:38:20 -04:00
2016-12-21 09:37:16 -05:00
if seg.IsEmpty() {
2016-12-02 15:40:58 -05:00
return
}
2016-07-03 16:14:38 -04:00
var maxack uint32
2016-11-13 16:27:58 -05:00
var maxackRemoved bool
2016-12-21 09:37:16 -05:00
for _, number := range seg.NumberList {
2017-02-26 09:01:50 -05:00
removed := v.processAck(number)
2016-07-03 16:14:38 -04:00
if maxack < number {
maxack = number
2016-11-13 16:27:58 -05:00
maxackRemoved = removed
2016-07-03 16:14:38 -04:00
}
}
2016-07-06 10:36:15 -04:00
2016-11-13 16:27:58 -05:00
if maxackRemoved {
2016-11-27 15:39:09 -05:00
v.window.HandleFastAck(maxack, rto)
2016-11-13 16:27:58 -05:00
if current-seg.Timestamp < 10000 {
2016-11-27 15:39:09 -05:00
v.conn.roundTrip.Update(current-seg.Timestamp, current)
2016-11-13 16:27:58 -05:00
}
}
2016-07-03 16:14:38 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWorker) Push(b []byte) int {
2016-07-03 16:14:38 -04:00
nBytes := 0
2016-11-27 15:39:09 -05:00
v.Lock()
defer v.Unlock()
2016-08-25 05:41:05 -04:00
2016-11-27 15:39:09 -05:00
for len(b) > 0 && !v.window.IsFull() {
2016-07-03 16:14:38 -04:00
var size int
2016-11-27 15:39:09 -05:00
if len(b) > int(v.conn.mss) {
size = int(v.conn.mss)
2016-07-03 16:14:38 -04:00
} else {
size = len(b)
}
2016-11-27 15:39:09 -05:00
v.window.Push(v.nextNumber, b[:size])
v.nextNumber++
2016-07-03 16:14:38 -04:00
b = b[size:]
nBytes += size
}
return nBytes
}
2016-08-24 05:17:42 -04:00
// Private: Visible for testing.
2016-12-20 16:53:58 -05:00
func (v *SendingWorker) Write(seg Segment) error {
2016-07-03 16:14:38 -04:00
dataSeg := seg.(*DataSegment)
2016-11-27 15:39:09 -05:00
dataSeg.Conv = v.conn.conv
dataSeg.SendingNext = v.firstUnacknowledged
2016-07-14 16:52:00 -04:00
dataSeg.Option = 0
2016-11-27 15:39:09 -05:00
if v.conn.State() == StateReadyToClose {
2016-07-14 16:52:00 -04:00
dataSeg.Option = SegmentOptionClose
2016-07-03 16:14:38 -04:00
}
2016-12-20 16:53:58 -05:00
return v.conn.output.Write(dataSeg)
2016-07-12 11:56:36 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWorker) OnPacketLoss(lossRate uint32) {
if !v.conn.Config.Congestion || v.conn.roundTrip.Timeout() == 0 {
2016-07-03 16:14:38 -04:00
return
}
2016-07-04 09:34:14 -04:00
if lossRate >= 15 {
2016-11-27 15:39:09 -05:00
v.controlWindow = 3 * v.controlWindow / 4
2016-07-04 09:34:14 -04:00
} else if lossRate <= 5 {
2016-11-27 15:39:09 -05:00
v.controlWindow += v.controlWindow / 4
2016-07-03 16:14:38 -04:00
}
2016-11-27 15:39:09 -05:00
if v.controlWindow < 16 {
v.controlWindow = 16
2016-07-03 16:14:38 -04:00
}
2016-11-27 15:39:09 -05:00
if v.controlWindow > 2*v.conn.Config.GetSendingInFlightSize() {
v.controlWindow = 2 * v.conn.Config.GetSendingInFlightSize()
2016-07-03 16:14:38 -04:00
}
}
2016-11-27 15:39:09 -05:00
func (v *SendingWorker) Flush(current uint32) {
v.Lock()
2016-07-03 16:14:38 -04:00
2016-11-27 15:39:09 -05:00
cwnd := v.firstUnacknowledged + v.conn.Config.GetSendingInFlightSize()
if cwnd > v.remoteNextNumber {
cwnd = v.remoteNextNumber
2016-07-03 16:14:38 -04:00
}
2016-11-27 15:39:09 -05:00
if v.conn.Config.Congestion && cwnd > v.firstUnacknowledged+v.controlWindow {
cwnd = v.firstUnacknowledged + v.controlWindow
2016-07-03 16:14:38 -04:00
}
2016-11-27 15:39:09 -05:00
if !v.window.IsEmpty() {
v.window.Flush(current, v.conn.roundTrip.Timeout(), cwnd)
2017-02-17 18:04:25 -05:00
v.firstUnacknowledgedUpdated = false
2016-08-24 09:47:14 -04:00
}
2016-10-11 06:24:19 -04:00
2017-02-17 18:04:25 -05:00
updated := v.firstUnacknowledgedUpdated
2016-11-27 15:39:09 -05:00
v.firstUnacknowledgedUpdated = false
2017-02-17 18:04:25 -05:00
v.Unlock()
if updated {
v.conn.Ping(current, CommandPing)
}
2016-07-03 16:14:38 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWorker) CloseWrite() {
v.Lock()
defer v.Unlock()
2016-07-03 16:14:38 -04:00
2016-11-27 15:39:09 -05:00
v.window.Clear(0xFFFFFFFF)
2016-07-03 16:14:38 -04:00
}
2016-07-12 17:54:54 -04:00
2016-11-27 15:39:09 -05:00
func (v *SendingWorker) IsEmpty() bool {
v.RLock()
defer v.RUnlock()
2016-07-12 17:54:54 -04:00
2016-11-27 15:39:09 -05:00
return v.window.IsEmpty()
2016-07-12 17:54:54 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *SendingWorker) UpdateNecessary() bool {
return !v.IsEmpty()
}
2017-02-17 18:04:25 -05:00
func (w *SendingWorker) FirstUnacknowledged() uint32 {
w.RLock()
defer w.RUnlock()
return w.firstUnacknowledged
}