1
0
mirror of https://github.com/v2fly/v2ray-core.git synced 2024-12-23 02:26:49 -05:00
v2fly/transport/internet/kcp/connection.go
2016-07-16 22:14:20 +02:00

516 lines
12 KiB
Go

package kcp
import (
"errors"
"io"
"net"
"sync"
"sync/atomic"
"time"
"github.com/v2ray/v2ray-core/common/alloc"
"github.com/v2ray/v2ray-core/common/log"
)
var (
errTimeout = errors.New("i/o timeout")
errBrokenPipe = errors.New("broken pipe")
errClosedListener = errors.New("Listener closed.")
errClosedConnection = errors.New("Connection closed.")
)
type State int32
const (
StateActive State = 0
StateReadyToClose State = 1
StatePeerClosed State = 2
StateTerminating State = 3
StatePeerTerminating State = 4
StateTerminated State = 5
)
const (
headerSize uint32 = 2
)
func nowMillisec() int64 {
now := time.Now()
return now.Unix()*1000 + int64(now.Nanosecond()/1000000)
}
type RountTripInfo struct {
sync.RWMutex
variation uint32
srtt uint32
rto uint32
minRtt uint32
}
func (this *RountTripInfo) Update(rtt uint32) {
if rtt > 0x7FFFFFFF {
return
}
this.Lock()
defer this.Unlock()
// https://tools.ietf.org/html/rfc6298
if this.srtt == 0 {
this.srtt = rtt
this.variation = rtt / 2
} else {
delta := rtt - this.srtt
if this.srtt > rtt {
delta = this.srtt - rtt
}
this.variation = (3*this.variation + delta) / 4
this.srtt = (7*this.srtt + rtt) / 8
if this.srtt < this.minRtt {
this.srtt = this.minRtt
}
}
var rto uint32
if this.minRtt < 4*this.variation {
rto = this.srtt + 4*this.variation
} else {
rto = this.srtt + this.variation
}
if rto > 10000 {
rto = 10000
}
this.rto = rto * 3 / 2
}
func (this *RountTripInfo) Timeout() uint32 {
this.RLock()
defer this.RUnlock()
return this.rto
}
func (this *RountTripInfo) SmoothedTime() uint32 {
this.RLock()
defer this.RUnlock()
return this.srtt
}
// Connection is a KCP connection over UDP.
type Connection struct {
block Authenticator
local, remote net.Addr
rd time.Time
wd time.Time // write deadline
writer io.WriteCloser
since int64
dataInputCond *sync.Cond
dataOutputCond *sync.Cond
conv uint16
state State
stateBeginTime uint32
lastIncomingTime uint32
lastPingTime uint32
mss uint32
roundTrip *RountTripInfo
interval uint32
receivingWorker *ReceivingWorker
sendingWorker *SendingWorker
fastresend uint32
congestionControl bool
output *BufferedSegmentWriter
}
// NewConnection create a new KCP connection between local and remote.
func NewConnection(conv uint16, writerCloser io.WriteCloser, local *net.UDPAddr, remote *net.UDPAddr, block Authenticator) *Connection {
log.Info("KCP|Connection: creating connection ", conv)
conn := new(Connection)
conn.local = local
conn.remote = remote
conn.block = block
conn.writer = writerCloser
conn.since = nowMillisec()
conn.dataInputCond = sync.NewCond(new(sync.Mutex))
conn.dataOutputCond = sync.NewCond(new(sync.Mutex))
authWriter := &AuthenticationWriter{
Authenticator: block,
Writer: writerCloser,
}
conn.conv = conv
conn.output = NewSegmentWriter(authWriter)
conn.mss = authWriter.Mtu() - DataSegmentOverhead
conn.roundTrip = &RountTripInfo{
rto: 100,
minRtt: effectiveConfig.Tti,
}
conn.interval = effectiveConfig.Tti
conn.receivingWorker = NewReceivingWorker(conn)
conn.fastresend = 2
conn.congestionControl = effectiveConfig.Congestion
conn.sendingWorker = NewSendingWorker(conn)
go conn.updateTask()
return conn
}
func (this *Connection) Elapsed() uint32 {
return uint32(nowMillisec() - this.since)
}
// Read implements the Conn Read method.
func (this *Connection) Read(b []byte) (int, error) {
if this == nil {
return 0, io.EOF
}
for {
if this.State() == StateReadyToClose || this.State() == StateTerminating || this.State() == StateTerminated {
return 0, io.EOF
}
nBytes := this.receivingWorker.Read(b)
if nBytes > 0 {
return nBytes, nil
}
if this.State() == StatePeerTerminating {
return 0, io.EOF
}
var timer *time.Timer
if !this.rd.IsZero() {
duration := this.rd.Sub(time.Now())
if duration <= 0 {
return 0, errTimeout
}
timer = time.AfterFunc(duration, this.dataInputCond.Signal)
}
this.dataInputCond.L.Lock()
this.dataInputCond.Wait()
this.dataInputCond.L.Unlock()
if timer != nil {
timer.Stop()
}
if !this.rd.IsZero() && this.rd.Before(time.Now()) {
return 0, errTimeout
}
}
}
// Write implements the Conn Write method.
func (this *Connection) Write(b []byte) (int, error) {
if this == nil || this.State() != StateActive {
return 0, io.ErrClosedPipe
}
totalWritten := 0
for {
if this == nil || this.State() != StateActive {
return totalWritten, io.ErrClosedPipe
}
nBytes := this.sendingWorker.Push(b[totalWritten:])
if nBytes > 0 {
totalWritten += nBytes
if totalWritten == len(b) {
return totalWritten, nil
}
}
var timer *time.Timer
if !this.wd.IsZero() {
duration := this.wd.Sub(time.Now())
if duration <= 0 {
return totalWritten, errTimeout
}
timer = time.AfterFunc(duration, this.dataOutputCond.Signal)
}
this.dataOutputCond.L.Lock()
this.dataOutputCond.Wait()
this.dataOutputCond.L.Unlock()
if timer != nil {
timer.Stop()
}
if !this.wd.IsZero() && this.wd.Before(time.Now()) {
return totalWritten, errTimeout
}
}
}
func (this *Connection) SetState(state State) {
current := this.Elapsed()
atomic.StoreInt32((*int32)(&this.state), int32(state))
atomic.StoreUint32(&this.stateBeginTime, current)
log.Debug("KCP|Connection: #", this.conv, " entering state ", state, " at ", current)
switch state {
case StateReadyToClose:
this.receivingWorker.CloseRead()
case StatePeerClosed:
this.sendingWorker.CloseWrite()
case StateTerminating:
this.receivingWorker.CloseRead()
this.sendingWorker.CloseWrite()
case StatePeerTerminating:
this.sendingWorker.CloseWrite()
case StateTerminated:
this.receivingWorker.CloseRead()
this.sendingWorker.CloseWrite()
}
}
// Close closes the connection.
func (this *Connection) Close() error {
if this == nil {
return errClosedConnection
}
this.dataInputCond.Broadcast()
this.dataOutputCond.Broadcast()
state := this.State()
if state == StateReadyToClose ||
state == StateTerminating ||
state == StateTerminated {
return errClosedConnection
}
log.Info("KCP|Connection: Closing connection to ", this.remote)
if state == StateActive {
this.SetState(StateReadyToClose)
}
if state == StatePeerClosed {
this.SetState(StateTerminating)
}
if state == StatePeerTerminating {
this.SetState(StateTerminated)
}
return nil
}
// LocalAddr returns the local network address. The Addr returned is shared by all invocations of LocalAddr, so do not modify it.
func (this *Connection) LocalAddr() net.Addr {
if this == nil {
return nil
}
return this.local
}
// RemoteAddr returns the remote network address. The Addr returned is shared by all invocations of RemoteAddr, so do not modify it.
func (this *Connection) RemoteAddr() net.Addr {
if this == nil {
return nil
}
return this.remote
}
// SetDeadline sets the deadline associated with the listener. A zero time value disables the deadline.
func (this *Connection) SetDeadline(t time.Time) error {
if err := this.SetReadDeadline(t); err != nil {
return err
}
if err := this.SetWriteDeadline(t); err != nil {
return err
}
return nil
}
// SetReadDeadline implements the Conn SetReadDeadline method.
func (this *Connection) SetReadDeadline(t time.Time) error {
if this == nil || this.State() != StateActive {
return errClosedConnection
}
this.rd = t
return nil
}
// SetWriteDeadline implements the Conn SetWriteDeadline method.
func (this *Connection) SetWriteDeadline(t time.Time) error {
if this == nil || this.State() != StateActive {
return errClosedConnection
}
this.wd = t
return nil
}
// kcp update, input loop
func (this *Connection) updateTask() {
for this.State() != StateTerminated {
this.flush()
interval := time.Duration(effectiveConfig.Tti) * time.Millisecond
if this.State() == StateTerminating {
interval = time.Second
}
time.Sleep(interval)
}
this.Terminate()
}
func (this *Connection) FetchInputFrom(conn io.Reader) {
go func() {
payload := alloc.NewLocalBuffer(2048)
defer payload.Release()
for {
payload.Reset()
nBytes, err := conn.Read(payload.Value)
if err != nil {
return
}
payload.Slice(0, nBytes)
if this.block.Open(payload) {
this.Input(payload.Value)
}
}
}()
}
func (this *Connection) Reusable() bool {
return false
}
func (this *Connection) SetReusable(b bool) {}
func (this *Connection) Terminate() {
if this == nil || this.writer == nil {
return
}
log.Info("KCP|Connection: Terminating connection to ", this.RemoteAddr())
this.writer.Close()
}
func (this *Connection) HandleOption(opt SegmentOption) {
if (opt & SegmentOptionClose) == SegmentOptionClose {
this.OnPeerClosed()
}
}
func (this *Connection) OnPeerClosed() {
state := this.State()
if state == StateReadyToClose {
this.SetState(StateTerminating)
}
if state == StateActive {
this.SetState(StatePeerClosed)
}
}
// Input when you received a low level packet (eg. UDP packet), call it
func (this *Connection) Input(data []byte) int {
current := this.Elapsed()
atomic.StoreUint32(&this.lastIncomingTime, current)
var seg Segment
for {
seg, data = ReadSegment(data)
if seg == nil {
break
}
switch seg := seg.(type) {
case *DataSegment:
this.HandleOption(seg.Option)
this.receivingWorker.ProcessSegment(seg)
this.dataInputCond.Signal()
case *AckSegment:
this.HandleOption(seg.Option)
this.sendingWorker.ProcessSegment(current, seg)
this.dataOutputCond.Signal()
case *CmdOnlySegment:
this.HandleOption(seg.Option)
if seg.Command == CommandTerminate {
state := this.State()
if state == StateActive ||
state == StatePeerClosed {
this.SetState(StatePeerTerminating)
} else if state == StateReadyToClose {
this.SetState(StateTerminating)
} else if state == StateTerminating {
this.SetState(StateTerminated)
}
}
this.sendingWorker.ProcessReceivingNext(seg.ReceivinNext)
this.receivingWorker.ProcessSendingNext(seg.SendingNext)
seg.Release()
default:
}
}
return 0
}
func (this *Connection) flush() {
current := this.Elapsed()
if this.State() == StateTerminated {
return
}
if this.State() == StateActive && current-atomic.LoadUint32(&this.lastIncomingTime) >= 30000 {
this.Close()
}
if this.State() == StateReadyToClose && this.sendingWorker.IsEmpty() {
this.SetState(StateTerminating)
}
if this.State() == StateTerminating {
log.Debug("KCP|Connection: #", this.conv, " sending terminating cmd.")
seg := NewCmdOnlySegment()
defer seg.Release()
seg.Conv = this.conv
seg.Command = CommandTerminate
this.output.Write(seg)
this.output.Flush()
if current-atomic.LoadUint32(&this.stateBeginTime) > 8000 {
this.SetState(StateTerminated)
}
return
}
if this.State() == StatePeerTerminating && current-atomic.LoadUint32(&this.stateBeginTime) > 4000 {
this.SetState(StateTerminating)
}
if this.State() == StateReadyToClose && current-atomic.LoadUint32(&this.stateBeginTime) > 15000 {
this.SetState(StateTerminating)
}
// flush acknowledges
this.receivingWorker.Flush(current)
this.sendingWorker.Flush(current)
if this.sendingWorker.PingNecessary() || this.receivingWorker.PingNecessary() || current-atomic.LoadUint32(&this.lastPingTime) >= 5000 {
seg := NewCmdOnlySegment()
seg.Conv = this.conv
seg.Command = CommandPing
seg.ReceivinNext = this.receivingWorker.nextNumber
seg.SendingNext = this.sendingWorker.firstUnacknowledged
if this.State() == StateReadyToClose {
seg.Option = SegmentOptionClose
}
this.output.Write(seg)
this.lastPingTime = current
this.sendingWorker.MarkPingNecessary(false)
this.receivingWorker.MarkPingNecessary(false)
seg.Release()
}
// flash remain segments
this.output.Flush()
}
func (this *Connection) State() State {
return State(atomic.LoadInt32((*int32)(&this.state)))
}