2016-06-25 15:35:18 -04:00
|
|
|
package kcp
|
|
|
|
|
2016-06-30 08:51:49 -04:00
|
|
|
import (
|
|
|
|
"io"
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/v2ray/v2ray-core/common/alloc"
|
|
|
|
)
|
|
|
|
|
2016-06-25 15:35:18 -04:00
|
|
|
type ReceivingWindow struct {
|
|
|
|
start uint32
|
|
|
|
size uint32
|
2016-06-29 04:34:34 -04:00
|
|
|
list []*DataSegment
|
2016-06-25 15:35:18 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
func NewReceivingWindow(size uint32) *ReceivingWindow {
|
|
|
|
return &ReceivingWindow{
|
|
|
|
start: 0,
|
|
|
|
size: size,
|
2016-06-29 04:34:34 -04:00
|
|
|
list: make([]*DataSegment, size),
|
2016-06-25 15:35:18 -04:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingWindow) Size() uint32 {
|
|
|
|
return this.size
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingWindow) Position(idx uint32) uint32 {
|
|
|
|
return (idx + this.start) % this.size
|
|
|
|
}
|
|
|
|
|
2016-06-29 04:34:34 -04:00
|
|
|
func (this *ReceivingWindow) Set(idx uint32, value *DataSegment) bool {
|
2016-06-25 15:35:18 -04:00
|
|
|
pos := this.Position(idx)
|
|
|
|
if this.list[pos] != nil {
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
this.list[pos] = value
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
2016-06-29 04:34:34 -04:00
|
|
|
func (this *ReceivingWindow) Remove(idx uint32) *DataSegment {
|
2016-06-25 15:35:18 -04:00
|
|
|
pos := this.Position(idx)
|
|
|
|
e := this.list[pos]
|
|
|
|
this.list[pos] = nil
|
|
|
|
return e
|
|
|
|
}
|
|
|
|
|
2016-06-29 04:34:34 -04:00
|
|
|
func (this *ReceivingWindow) RemoveFirst() *DataSegment {
|
2016-06-25 15:35:18 -04:00
|
|
|
return this.Remove(0)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingWindow) Advance() {
|
|
|
|
this.start++
|
|
|
|
if this.start == this.size {
|
|
|
|
this.start = 0
|
|
|
|
}
|
|
|
|
}
|
2016-06-27 16:34:46 -04:00
|
|
|
|
2016-06-30 08:51:49 -04:00
|
|
|
type ReceivingQueue struct {
|
2016-07-02 16:28:28 -04:00
|
|
|
sync.Mutex
|
2016-06-30 08:51:49 -04:00
|
|
|
closed bool
|
|
|
|
cache *alloc.Buffer
|
|
|
|
queue chan *alloc.Buffer
|
|
|
|
timeout time.Time
|
|
|
|
}
|
|
|
|
|
2016-07-02 17:37:51 -04:00
|
|
|
func NewReceivingQueue(size uint32) *ReceivingQueue {
|
2016-06-30 08:51:49 -04:00
|
|
|
return &ReceivingQueue{
|
2016-07-02 17:37:51 -04:00
|
|
|
queue: make(chan *alloc.Buffer, size),
|
2016-06-30 08:51:49 -04:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingQueue) Read(buf []byte) (int, error) {
|
2016-07-02 17:37:51 -04:00
|
|
|
if this.closed {
|
|
|
|
return 0, io.EOF
|
|
|
|
}
|
|
|
|
|
2016-06-30 08:51:49 -04:00
|
|
|
if this.cache.Len() > 0 {
|
|
|
|
nBytes, err := this.cache.Read(buf)
|
|
|
|
if this.cache.IsEmpty() {
|
|
|
|
this.cache.Release()
|
|
|
|
this.cache = nil
|
|
|
|
}
|
|
|
|
return nBytes, err
|
|
|
|
}
|
|
|
|
|
|
|
|
var totalBytes int
|
|
|
|
|
|
|
|
L:
|
|
|
|
for totalBytes < len(buf) {
|
|
|
|
timeToSleep := time.Millisecond
|
|
|
|
select {
|
|
|
|
case payload, open := <-this.queue:
|
|
|
|
if !open {
|
|
|
|
return totalBytes, io.EOF
|
|
|
|
}
|
|
|
|
nBytes, err := payload.Read(buf)
|
|
|
|
totalBytes += nBytes
|
|
|
|
if err != nil {
|
|
|
|
return totalBytes, err
|
|
|
|
}
|
|
|
|
if !payload.IsEmpty() {
|
|
|
|
this.cache = payload
|
|
|
|
}
|
|
|
|
buf = buf[nBytes:]
|
|
|
|
case <-time.After(timeToSleep):
|
|
|
|
if totalBytes > 0 {
|
|
|
|
break L
|
|
|
|
}
|
|
|
|
if !this.timeout.IsZero() && this.timeout.Before(time.Now()) {
|
|
|
|
return totalBytes, errTimeout
|
|
|
|
}
|
|
|
|
timeToSleep += 500 * time.Millisecond
|
2016-07-05 08:08:08 -04:00
|
|
|
if timeToSleep > 5*time.Second {
|
|
|
|
timeToSleep = 5 * time.Second
|
|
|
|
}
|
2016-06-30 08:51:49 -04:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return totalBytes, nil
|
|
|
|
}
|
|
|
|
|
2016-07-02 15:26:50 -04:00
|
|
|
func (this *ReceivingQueue) Put(payload *alloc.Buffer) bool {
|
2016-06-30 10:58:38 -04:00
|
|
|
if this.closed {
|
|
|
|
payload.Release()
|
2016-07-02 15:26:50 -04:00
|
|
|
return false
|
2016-06-30 10:58:38 -04:00
|
|
|
}
|
|
|
|
|
2016-07-02 15:26:50 -04:00
|
|
|
select {
|
|
|
|
case this.queue <- payload:
|
|
|
|
return true
|
|
|
|
default:
|
|
|
|
return false
|
|
|
|
}
|
2016-06-30 08:51:49 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingQueue) SetReadDeadline(t time.Time) error {
|
|
|
|
this.timeout = t
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingQueue) Close() {
|
2016-07-02 16:28:28 -04:00
|
|
|
this.Lock()
|
|
|
|
defer this.Unlock()
|
|
|
|
|
2016-06-30 08:51:49 -04:00
|
|
|
if this.closed {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
this.closed = true
|
|
|
|
close(this.queue)
|
|
|
|
}
|
|
|
|
|
2016-07-02 05:33:34 -04:00
|
|
|
type AckList struct {
|
2016-07-02 16:28:28 -04:00
|
|
|
sync.Mutex
|
2016-07-02 15:26:50 -04:00
|
|
|
writer SegmentWriter
|
2016-06-27 16:34:46 -04:00
|
|
|
timestamps []uint32
|
|
|
|
numbers []uint32
|
2016-06-30 16:19:30 -04:00
|
|
|
nextFlush []uint32
|
|
|
|
}
|
|
|
|
|
2016-07-02 17:18:12 -04:00
|
|
|
func NewAckList(writer SegmentWriter) *AckList {
|
2016-07-02 05:33:34 -04:00
|
|
|
return &AckList{
|
2016-07-02 15:26:50 -04:00
|
|
|
writer: writer,
|
2016-06-30 16:19:30 -04:00
|
|
|
timestamps: make([]uint32, 0, 32),
|
|
|
|
numbers: make([]uint32, 0, 32),
|
|
|
|
nextFlush: make([]uint32, 0, 32),
|
|
|
|
}
|
2016-06-27 16:34:46 -04:00
|
|
|
}
|
|
|
|
|
2016-07-02 05:33:34 -04:00
|
|
|
func (this *AckList) Add(number uint32, timestamp uint32) {
|
2016-07-02 16:28:28 -04:00
|
|
|
this.Lock()
|
|
|
|
defer this.Unlock()
|
|
|
|
|
2016-06-27 16:34:46 -04:00
|
|
|
this.timestamps = append(this.timestamps, timestamp)
|
|
|
|
this.numbers = append(this.numbers, number)
|
2016-06-30 16:19:30 -04:00
|
|
|
this.nextFlush = append(this.nextFlush, 0)
|
2016-06-27 16:34:46 -04:00
|
|
|
}
|
|
|
|
|
2016-07-02 05:33:34 -04:00
|
|
|
func (this *AckList) Clear(una uint32) {
|
2016-07-02 16:28:28 -04:00
|
|
|
this.Lock()
|
|
|
|
defer this.Unlock()
|
|
|
|
|
2016-06-27 16:34:46 -04:00
|
|
|
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]
|
2016-06-30 16:19:30 -04:00
|
|
|
this.nextFlush[count] = this.nextFlush[i]
|
2016-06-27 16:34:46 -04:00
|
|
|
}
|
|
|
|
count++
|
|
|
|
}
|
|
|
|
}
|
2016-06-29 17:41:04 -04:00
|
|
|
if count < len(this.numbers) {
|
|
|
|
this.numbers = this.numbers[:count]
|
|
|
|
this.timestamps = this.timestamps[:count]
|
2016-06-30 16:19:30 -04:00
|
|
|
this.nextFlush = this.nextFlush[:count]
|
2016-06-29 17:41:04 -04:00
|
|
|
}
|
2016-06-27 16:34:46 -04:00
|
|
|
}
|
|
|
|
|
2016-07-02 16:28:28 -04:00
|
|
|
func (this *AckList) Flush(current uint32, rto uint32) {
|
2016-07-05 04:28:23 -04:00
|
|
|
seg := NewAckSegment()
|
2016-07-02 16:28:28 -04:00
|
|
|
this.Lock()
|
2016-07-05 04:33:11 -04:00
|
|
|
for i := 0; i < len(this.numbers) && !seg.IsFull(); i++ {
|
2016-06-30 16:19:30 -04:00
|
|
|
if this.nextFlush[i] <= current {
|
2016-07-05 04:33:11 -04:00
|
|
|
seg.PutNumber(this.numbers[i], this.timestamps[i])
|
2016-07-02 16:28:28 -04:00
|
|
|
this.nextFlush[i] = current + rto/2
|
2016-06-30 16:19:30 -04:00
|
|
|
}
|
2016-06-27 16:34:46 -04:00
|
|
|
}
|
2016-07-02 16:28:28 -04:00
|
|
|
this.Unlock()
|
2016-06-30 16:19:30 -04:00
|
|
|
if seg.Count > 0 {
|
2016-07-02 15:26:50 -04:00
|
|
|
this.writer.Write(seg)
|
2016-07-05 08:08:08 -04:00
|
|
|
seg.Release()
|
2016-06-30 16:19:30 -04:00
|
|
|
}
|
2016-07-02 15:26:50 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
type ReceivingWorker struct {
|
2016-07-02 16:28:28 -04:00
|
|
|
kcp *KCP
|
|
|
|
queue *ReceivingQueue
|
|
|
|
window *ReceivingWindow
|
|
|
|
windowMutex sync.Mutex
|
|
|
|
acklist *AckList
|
|
|
|
updated bool
|
|
|
|
nextNumber uint32
|
|
|
|
windowSize uint32
|
2016-07-02 15:26:50 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
func NewReceivingWorker(kcp *KCP) *ReceivingWorker {
|
|
|
|
windowSize := effectiveConfig.GetReceivingWindowSize()
|
|
|
|
worker := &ReceivingWorker{
|
|
|
|
kcp: kcp,
|
2016-07-02 17:37:51 -04:00
|
|
|
queue: NewReceivingQueue(effectiveConfig.GetReceivingQueueSize()),
|
2016-07-02 15:26:50 -04:00
|
|
|
window: NewReceivingWindow(windowSize),
|
|
|
|
windowSize: windowSize,
|
|
|
|
}
|
2016-07-02 17:18:12 -04:00
|
|
|
worker.acklist = NewAckList(worker)
|
2016-07-02 15:26:50 -04:00
|
|
|
return worker
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingWorker) ProcessSendingNext(number uint32) {
|
|
|
|
this.acklist.Clear(number)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingWorker) ProcessSegment(seg *DataSegment) {
|
|
|
|
number := seg.Number
|
2016-07-02 18:38:33 -04:00
|
|
|
idx := number - this.nextNumber
|
|
|
|
if idx >= this.windowSize {
|
2016-07-02 15:26:50 -04:00
|
|
|
return
|
|
|
|
}
|
|
|
|
this.ProcessSendingNext(seg.SendingNext)
|
|
|
|
this.acklist.Add(number, seg.Timestamp)
|
2016-07-02 16:28:28 -04:00
|
|
|
this.windowMutex.Lock()
|
2016-07-03 03:38:51 -04:00
|
|
|
defer this.windowMutex.Unlock()
|
2016-07-02 15:26:50 -04:00
|
|
|
|
|
|
|
if !this.window.Set(idx, seg) {
|
|
|
|
seg.Release()
|
|
|
|
}
|
|
|
|
|
|
|
|
for {
|
|
|
|
seg := this.window.RemoveFirst()
|
|
|
|
if seg == nil {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
if !this.queue.Put(seg.Data) {
|
|
|
|
this.window.Set(0, seg)
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
seg.Data = nil
|
2016-07-05 08:08:08 -04:00
|
|
|
seg.Release()
|
2016-07-02 15:26:50 -04:00
|
|
|
this.window.Advance()
|
|
|
|
this.nextNumber++
|
|
|
|
this.updated = true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingWorker) Read(b []byte) (int, error) {
|
|
|
|
return this.queue.Read(b)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingWorker) SetReadDeadline(t time.Time) {
|
|
|
|
this.queue.SetReadDeadline(t)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingWorker) Flush() {
|
2016-07-02 16:28:28 -04:00
|
|
|
this.acklist.Flush(this.kcp.current, this.kcp.rx_rto)
|
2016-07-02 15:26:50 -04:00
|
|
|
}
|
|
|
|
|
2016-07-04 08:17:42 -04:00
|
|
|
func (this *ReceivingWorker) Write(seg Segment) {
|
2016-07-02 16:17:41 -04:00
|
|
|
ackSeg := seg.(*AckSegment)
|
2016-07-02 15:26:50 -04:00
|
|
|
ackSeg.Conv = this.kcp.conv
|
|
|
|
ackSeg.ReceivingNext = this.nextNumber
|
|
|
|
ackSeg.ReceivingWindow = this.nextNumber + this.windowSize
|
|
|
|
if this.kcp.state == StateReadyToClose {
|
|
|
|
ackSeg.Opt = SegmentOptionClose
|
|
|
|
}
|
|
|
|
this.kcp.output.Write(ackSeg)
|
|
|
|
this.updated = false
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingWorker) CloseRead() {
|
|
|
|
this.queue.Close()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (this *ReceivingWorker) PingNecessary() bool {
|
|
|
|
return this.updated
|
2016-06-27 16:34:46 -04:00
|
|
|
}
|