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

586 lines
13 KiB
Go
Raw Normal View History

2016-06-17 10:51:41 -04:00
package kcp
import (
"io"
"net"
"sync"
2016-07-05 17:53:46 -04:00
"sync/atomic"
2016-06-17 10:51:41 -04:00
"time"
2016-12-08 10:27:41 -05:00
2016-12-04 03:10:47 -05:00
"v2ray.com/core/common/errors"
2016-08-20 14:55:45 -04:00
"v2ray.com/core/common/log"
"v2ray.com/core/common/predicate"
2016-11-27 02:58:31 -05:00
"v2ray.com/core/transport/internet/internal"
2016-06-17 10:51:41 -04:00
)
var (
2016-08-07 09:31:24 -04:00
ErrIOTimeout = errors.New("Read/Write timeout")
ErrClosedListener = errors.New("Listener closed.")
ErrClosedConnection = errors.New("Connection closed.")
2016-06-17 10:51:41 -04:00
)
2016-07-05 17:53:46 -04:00
type State int32
2016-07-05 17:02:52 -04:00
2016-11-27 15:39:09 -05:00
func (v State) Is(states ...State) bool {
2016-07-26 15:34:00 -04:00
for _, state := range states {
2016-11-27 15:39:09 -05:00
if v == state {
2016-07-26 15:34:00 -04:00
return true
}
}
return false
}
2016-07-05 17:02:52 -04:00
const (
StateActive State = 0
StateReadyToClose State = 1
StatePeerClosed State = 2
StateTerminating State = 3
StatePeerTerminating State = 4
StateTerminated State = 5
2016-07-05 17:02:52 -04:00
)
2016-06-17 10:51:41 -04:00
func nowMillisec() int64 {
now := time.Now()
return now.Unix()*1000 + int64(now.Nanosecond()/1000000)
}
2016-08-05 15:03:44 -04:00
type RoundTripInfo struct {
2016-07-05 17:53:46 -04:00
sync.RWMutex
variation uint32
srtt uint32
rto uint32
minRtt uint32
updatedTimestamp uint32
2016-07-05 17:53:46 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *RoundTripInfo) UpdatePeerRTO(rto uint32, current uint32) {
v.Lock()
defer v.Unlock()
2016-11-27 15:39:09 -05:00
if current-v.updatedTimestamp < 3000 {
return
}
2016-11-27 15:39:09 -05:00
v.updatedTimestamp = current
v.rto = rto
}
2016-11-27 15:39:09 -05:00
func (v *RoundTripInfo) Update(rtt uint32, current uint32) {
2016-07-05 17:53:46 -04:00
if rtt > 0x7FFFFFFF {
return
}
2016-11-27 15:39:09 -05:00
v.Lock()
defer v.Unlock()
2016-07-05 17:53:46 -04:00
// https://tools.ietf.org/html/rfc6298
2016-11-27 15:39:09 -05:00
if v.srtt == 0 {
v.srtt = rtt
v.variation = rtt / 2
2016-07-05 17:53:46 -04:00
} else {
2016-11-27 15:39:09 -05:00
delta := rtt - v.srtt
if v.srtt > rtt {
delta = v.srtt - rtt
2016-07-05 17:53:46 -04:00
}
2016-11-27 15:39:09 -05:00
v.variation = (3*v.variation + delta) / 4
v.srtt = (7*v.srtt + rtt) / 8
if v.srtt < v.minRtt {
v.srtt = v.minRtt
2016-07-05 17:53:46 -04:00
}
}
var rto uint32
2016-11-27 15:39:09 -05:00
if v.minRtt < 4*v.variation {
rto = v.srtt + 4*v.variation
2016-07-05 17:53:46 -04:00
} else {
2016-11-27 15:39:09 -05:00
rto = v.srtt + v.variation
2016-07-05 17:53:46 -04:00
}
if rto > 10000 {
rto = 10000
}
2016-11-27 15:39:09 -05:00
v.rto = rto * 5 / 4
v.updatedTimestamp = current
2016-07-05 17:53:46 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *RoundTripInfo) Timeout() uint32 {
v.RLock()
defer v.RUnlock()
2016-07-05 17:53:46 -04:00
2016-11-27 15:39:09 -05:00
return v.rto
2016-07-05 17:53:46 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *RoundTripInfo) SmoothedTime() uint32 {
v.RLock()
defer v.RUnlock()
2016-07-05 17:53:46 -04:00
2016-11-27 15:39:09 -05:00
return v.srtt
2016-07-05 17:53:46 -04:00
}
type Updater struct {
interval time.Duration
shouldContinue predicate.Predicate
shouldTerminate predicate.Predicate
updateFunc func()
notifier chan bool
}
func NewUpdater(interval uint32, shouldContinue predicate.Predicate, shouldTerminate predicate.Predicate, updateFunc func()) *Updater {
u := &Updater{
interval: time.Duration(interval) * time.Millisecond,
shouldContinue: shouldContinue,
shouldTerminate: shouldTerminate,
updateFunc: updateFunc,
notifier: make(chan bool, 1),
}
go u.Run()
return u
}
2016-11-27 15:39:09 -05:00
func (v *Updater) WakeUp() {
select {
2016-11-27 15:39:09 -05:00
case v.notifier <- true:
default:
}
}
2016-11-27 15:39:09 -05:00
func (v *Updater) Run() {
for <-v.notifier {
if v.shouldTerminate() {
return
}
2016-11-27 15:39:09 -05:00
for v.shouldContinue() {
v.updateFunc()
time.Sleep(v.interval)
}
}
}
2016-11-27 02:58:31 -05:00
type SystemConnection interface {
net.Conn
2017-01-01 16:12:44 -05:00
Id() internal.ConnectionID
2016-12-08 10:27:41 -05:00
Reset(func([]Segment))
Overhead() int
2016-11-27 02:58:31 -05:00
}
2016-06-18 10:34:04 -04:00
// Connection is a KCP connection over UDP.
type Connection struct {
2016-12-08 16:50:12 -05:00
conn SystemConnection
connRecycler internal.ConnectionRecyler
rd time.Time
wd time.Time // write deadline
since int64
dataInput chan bool
dataOutput chan bool
Config *Config
2016-07-05 17:02:52 -04:00
conv uint16
state State
stateBeginTime uint32
lastIncomingTime uint32
lastPingTime uint32
2016-07-05 17:53:46 -04:00
mss uint32
2016-08-05 15:03:44 -04:00
roundTrip *RoundTripInfo
2016-07-05 17:02:52 -04:00
receivingWorker *ReceivingWorker
sendingWorker *SendingWorker
2016-12-20 16:53:58 -05:00
output SegmentWriter
dataUpdater *Updater
pingUpdater *Updater
2016-11-27 02:58:31 -05:00
reusable bool
2016-06-17 10:51:41 -04:00
}
2016-06-18 10:34:04 -04:00
// NewConnection create a new KCP connection between local and remote.
2016-12-08 10:27:41 -05:00
func NewConnection(conv uint16, sysConn SystemConnection, recycler internal.ConnectionRecyler, config *Config) *Connection {
2016-07-11 11:24:35 -04:00
log.Info("KCP|Connection: creating connection ", conv)
2016-07-05 17:02:52 -04:00
2016-11-27 02:58:31 -05:00
conn := &Connection{
2016-12-08 16:50:12 -05:00
conv: conv,
conn: sysConn,
connRecycler: recycler,
since: nowMillisec(),
dataInput: make(chan bool, 1),
dataOutput: make(chan bool, 1),
Config: config,
2016-12-20 16:53:58 -05:00
output: NewSegmentWriter(sysConn),
2016-12-23 06:42:25 -05:00
mss: config.GetMTUValue() - uint32(sysConn.Overhead()) - DataSegmentOverhead,
2016-11-27 02:58:31 -05:00
roundTrip: &RoundTripInfo{
rto: 100,
2016-12-23 06:42:25 -05:00
minRtt: config.GetTTIValue(),
2016-11-27 02:58:31 -05:00
},
}
2016-12-08 10:27:41 -05:00
sysConn.Reset(conn.Input)
2016-11-27 02:58:31 -05:00
2016-07-05 17:02:52 -04:00
conn.receivingWorker = NewReceivingWorker(conn)
conn.sendingWorker = NewSendingWorker(conn)
2016-06-17 10:51:41 -04:00
2016-10-11 06:30:53 -04:00
isTerminating := func() bool {
return conn.State().Is(StateTerminating, StateTerminated)
}
isTerminated := func() bool {
return conn.State() == StateTerminated
}
conn.dataUpdater = NewUpdater(
2016-12-23 06:42:25 -05:00
config.GetTTIValue(),
2016-10-11 06:44:17 -04:00
predicate.Not(isTerminating).And(predicate.Any(conn.sendingWorker.UpdateNecessary, conn.receivingWorker.UpdateNecessary)),
2016-10-11 06:30:53 -04:00
isTerminating,
conn.updateTask)
conn.pingUpdater = NewUpdater(
2016-10-11 06:24:19 -04:00
5000, // 5 seconds
2016-10-11 06:30:53 -04:00
predicate.Not(isTerminated),
isTerminated,
conn.updateTask)
2016-10-11 06:24:19 -04:00
conn.pingUpdater.WakeUp()
2016-06-17 10:51:41 -04:00
2016-06-18 10:34:04 -04:00
return conn
2016-06-17 10:51:41 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *Connection) Elapsed() uint32 {
return uint32(nowMillisec() - v.since)
2016-06-17 10:51:41 -04:00
}
2016-12-08 16:50:12 -05:00
func (v *Connection) OnDataInput() {
select {
case v.dataInput <- true:
default:
}
}
func (v *Connection) OnDataOutput() {
select {
case v.dataOutput <- true:
default:
}
}
2016-06-17 10:51:41 -04:00
// Read implements the Conn Read method.
2016-11-27 15:39:09 -05:00
func (v *Connection) Read(b []byte) (int, error) {
if v == nil {
2016-06-17 10:51:41 -04:00
return 0, io.EOF
}
2016-07-05 17:02:52 -04:00
2016-07-06 11:34:38 -04:00
for {
2016-11-27 15:39:09 -05:00
if v.State().Is(StateReadyToClose, StateTerminating, StateTerminated) {
2016-07-06 11:34:38 -04:00
return 0, io.EOF
}
2016-11-27 15:39:09 -05:00
nBytes := v.receivingWorker.Read(b)
2016-07-06 11:34:38 -04:00
if nBytes > 0 {
return nBytes, nil
}
2016-07-12 17:54:54 -04:00
2016-11-27 15:39:09 -05:00
if v.State() == StatePeerTerminating {
return 0, io.EOF
}
2016-12-24 18:01:24 -05:00
duration := time.Minute
2016-11-27 15:39:09 -05:00
if !v.rd.IsZero() {
2016-12-08 16:50:12 -05:00
duration = v.rd.Sub(time.Now())
if duration < 0 {
2016-08-07 09:31:24 -04:00
return 0, ErrIOTimeout
2016-07-12 06:54:09 -04:00
}
2016-07-06 11:34:38 -04:00
}
2016-12-08 16:50:12 -05:00
select {
case <-v.dataInput:
case <-time.After(duration):
if !v.rd.IsZero() && v.rd.Before(time.Now()) {
return 0, ErrIOTimeout
}
2016-07-06 11:34:38 -04:00
}
2016-07-05 17:02:52 -04:00
}
2016-06-17 10:51:41 -04:00
}
// Write implements the Conn Write method.
2016-11-27 15:39:09 -05:00
func (v *Connection) Write(b []byte) (int, error) {
2016-06-26 17:51:17 -04:00
totalWritten := 0
2016-06-17 10:51:41 -04:00
for {
2016-11-27 15:39:09 -05:00
if v == nil || v.State() != StateActive {
2016-06-26 17:51:17 -04:00
return totalWritten, io.ErrClosedPipe
2016-06-17 10:51:41 -04:00
}
2016-11-27 15:39:09 -05:00
nBytes := v.sendingWorker.Push(b[totalWritten:])
v.dataUpdater.WakeUp()
2016-06-26 17:51:17 -04:00
if nBytes > 0 {
totalWritten += nBytes
if totalWritten == len(b) {
return totalWritten, nil
}
2016-06-17 10:51:41 -04:00
}
2016-06-26 17:51:17 -04:00
2016-12-24 18:01:24 -05:00
duration := time.Minute
2016-12-24 16:04:09 -05:00
if !v.wd.IsZero() {
2016-12-08 16:50:12 -05:00
duration = v.wd.Sub(time.Now())
if duration < 0 {
2016-08-07 09:31:24 -04:00
return totalWritten, ErrIOTimeout
2016-07-14 11:39:12 -04:00
}
}
2016-12-08 16:50:12 -05:00
select {
case <-v.dataOutput:
case <-time.After(duration):
if !v.wd.IsZero() && v.wd.Before(time.Now()) {
return totalWritten, ErrIOTimeout
}
2016-06-17 10:51:41 -04:00
}
}
}
2016-11-27 15:39:09 -05:00
func (v *Connection) SetState(state State) {
current := v.Elapsed()
atomic.StoreInt32((*int32)(&v.state), int32(state))
atomic.StoreUint32(&v.stateBeginTime, current)
log.Debug("KCP|Connection: #", v.conv, " entering state ", state, " at ", current)
2016-07-05 17:02:52 -04:00
switch state {
case StateReadyToClose:
2016-11-27 15:39:09 -05:00
v.receivingWorker.CloseRead()
2016-07-05 17:02:52 -04:00
case StatePeerClosed:
2016-11-27 15:39:09 -05:00
v.sendingWorker.CloseWrite()
2016-07-05 17:02:52 -04:00
case StateTerminating:
2016-11-27 15:39:09 -05:00
v.receivingWorker.CloseRead()
v.sendingWorker.CloseWrite()
v.pingUpdater.interval = time.Second
case StatePeerTerminating:
2016-11-27 15:39:09 -05:00
v.sendingWorker.CloseWrite()
v.pingUpdater.interval = time.Second
2016-07-05 17:02:52 -04:00
case StateTerminated:
2016-11-27 15:39:09 -05:00
v.receivingWorker.CloseRead()
v.sendingWorker.CloseWrite()
v.pingUpdater.interval = time.Second
v.dataUpdater.WakeUp()
v.pingUpdater.WakeUp()
go v.Terminate()
2016-07-05 17:02:52 -04:00
}
}
2016-06-17 10:51:41 -04:00
// Close closes the connection.
2016-11-27 15:39:09 -05:00
func (v *Connection) Close() error {
if v == nil {
2016-08-07 09:31:24 -04:00
return ErrClosedConnection
2016-07-05 17:02:52 -04:00
}
2016-12-08 16:50:12 -05:00
v.OnDataInput()
v.OnDataOutput()
2016-07-06 11:34:38 -04:00
2016-11-27 15:39:09 -05:00
state := v.State()
2016-07-26 15:34:00 -04:00
if state.Is(StateReadyToClose, StateTerminating, StateTerminated) {
2016-08-07 09:31:24 -04:00
return ErrClosedConnection
2016-06-17 17:50:08 -04:00
}
2016-11-27 15:39:09 -05:00
log.Info("KCP|Connection: Closing connection to ", v.conn.RemoteAddr())
2016-06-18 10:34:04 -04:00
2016-07-05 17:02:52 -04:00
if state == StateActive {
2016-11-27 15:39:09 -05:00
v.SetState(StateReadyToClose)
2016-07-05 17:02:52 -04:00
}
if state == StatePeerClosed {
2016-11-27 15:39:09 -05:00
v.SetState(StateTerminating)
2016-07-05 17:02:52 -04:00
}
if state == StatePeerTerminating {
2016-11-27 15:39:09 -05:00
v.SetState(StateTerminated)
}
2016-06-17 10:51:41 -04:00
return nil
}
// LocalAddr returns the local network address. The Addr returned is shared by all invocations of LocalAddr, so do not modify it.
2016-11-27 15:39:09 -05:00
func (v *Connection) LocalAddr() net.Addr {
if v == nil {
2016-06-17 17:50:08 -04:00
return nil
}
2016-11-27 15:39:09 -05:00
return v.conn.LocalAddr()
2016-06-17 10:51:41 -04:00
}
// RemoteAddr returns the remote network address. The Addr returned is shared by all invocations of RemoteAddr, so do not modify it.
2016-11-27 15:39:09 -05:00
func (v *Connection) RemoteAddr() net.Addr {
if v == nil {
2016-06-17 17:50:08 -04:00
return nil
}
2016-11-27 15:39:09 -05:00
return v.conn.RemoteAddr()
2016-06-17 17:50:08 -04:00
}
2016-06-17 10:51:41 -04:00
// SetDeadline sets the deadline associated with the listener. A zero time value disables the deadline.
2016-11-27 15:39:09 -05:00
func (v *Connection) SetDeadline(t time.Time) error {
if err := v.SetReadDeadline(t); err != nil {
2016-06-30 08:51:49 -04:00
return err
}
2016-11-27 15:39:09 -05:00
if err := v.SetWriteDeadline(t); err != nil {
2016-06-30 08:51:49 -04:00
return err
2016-06-17 17:50:08 -04:00
}
2016-06-17 10:51:41 -04:00
return nil
}
// SetReadDeadline implements the Conn SetReadDeadline method.
2016-11-27 15:39:09 -05:00
func (v *Connection) SetReadDeadline(t time.Time) error {
if v == nil || v.State() != StateActive {
2016-08-07 09:31:24 -04:00
return ErrClosedConnection
2016-06-17 17:50:08 -04:00
}
2016-11-27 15:39:09 -05:00
v.rd = t
2016-06-17 10:51:41 -04:00
return nil
}
// SetWriteDeadline implements the Conn SetWriteDeadline method.
2016-11-27 15:39:09 -05:00
func (v *Connection) SetWriteDeadline(t time.Time) error {
if v == nil || v.State() != StateActive {
2016-08-07 09:31:24 -04:00
return ErrClosedConnection
2016-06-17 17:50:08 -04:00
}
2016-11-27 15:39:09 -05:00
v.wd = t
2016-06-17 10:51:41 -04:00
return nil
}
// kcp update, input loop
2016-11-27 15:39:09 -05:00
func (v *Connection) updateTask() {
v.flush()
2016-06-17 10:51:41 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *Connection) Reusable() bool {
2016-12-23 06:42:25 -05:00
return v.Config.IsConnectionReuse() && v.reusable
2016-06-17 10:51:41 -04:00
}
2016-11-27 15:39:09 -05:00
func (v *Connection) SetReusable(b bool) {
v.reusable = b
2016-11-27 03:02:27 -05:00
}
2016-06-29 04:34:34 -04:00
2016-11-27 15:39:09 -05:00
func (v *Connection) Terminate() {
if v == nil {
2016-06-29 04:34:34 -04:00
return
}
2016-11-27 15:39:09 -05:00
log.Info("KCP|Connection: Terminating connection to ", v.RemoteAddr())
2016-06-29 04:34:34 -04:00
2016-11-27 15:39:09 -05:00
//v.SetState(StateTerminated)
2016-12-08 16:50:12 -05:00
v.OnDataInput()
v.OnDataOutput()
2016-12-23 06:42:25 -05:00
if v.Config.IsConnectionReuse() && v.reusable {
2016-11-27 15:39:09 -05:00
v.connRecycler.Put(v.conn.Id(), v.conn)
2016-11-27 02:58:31 -05:00
} else {
2016-11-27 15:39:09 -05:00
v.conn.Close()
2016-11-27 02:58:31 -05:00
}
2016-11-27 15:39:09 -05:00
v.sendingWorker.Release()
v.receivingWorker.Release()
2016-06-29 04:34:34 -04:00
}
2016-07-05 17:02:52 -04:00
2016-11-27 15:39:09 -05:00
func (v *Connection) HandleOption(opt SegmentOption) {
2016-07-05 17:02:52 -04:00
if (opt & SegmentOptionClose) == SegmentOptionClose {
2016-11-27 15:39:09 -05:00
v.OnPeerClosed()
2016-07-05 17:02:52 -04:00
}
}
2016-11-27 15:39:09 -05:00
func (v *Connection) OnPeerClosed() {
state := v.State()
2016-07-05 17:02:52 -04:00
if state == StateReadyToClose {
2016-11-27 15:39:09 -05:00
v.SetState(StateTerminating)
2016-07-05 17:02:52 -04:00
}
if state == StateActive {
2016-11-27 15:39:09 -05:00
v.SetState(StatePeerClosed)
2016-07-05 17:02:52 -04:00
}
}
// Input when you received a low level packet (eg. UDP packet), call it
2016-12-08 10:27:41 -05:00
func (v *Connection) Input(segments []Segment) {
2016-11-27 15:39:09 -05:00
current := v.Elapsed()
atomic.StoreUint32(&v.lastIncomingTime, current)
2016-07-05 17:02:52 -04:00
2016-12-08 10:27:41 -05:00
for _, seg := range segments {
2016-11-27 15:39:09 -05:00
if seg.Conversation() != v.conv {
2016-12-08 16:50:12 -05:00
break
2016-11-27 02:58:31 -05:00
}
2016-07-05 17:02:52 -04:00
switch seg := seg.(type) {
case *DataSegment:
2016-11-27 15:39:09 -05:00
v.HandleOption(seg.Option)
v.receivingWorker.ProcessSegment(seg)
2016-12-08 16:50:12 -05:00
if seg.Number == v.receivingWorker.nextNumber {
v.OnDataInput()
}
2016-11-27 15:39:09 -05:00
v.dataUpdater.WakeUp()
2016-07-05 17:02:52 -04:00
case *AckSegment:
2016-11-27 15:39:09 -05:00
v.HandleOption(seg.Option)
v.sendingWorker.ProcessSegment(current, seg, v.roundTrip.Timeout())
2016-12-08 16:50:12 -05:00
v.OnDataOutput()
2016-11-27 15:39:09 -05:00
v.dataUpdater.WakeUp()
2016-07-05 17:02:52 -04:00
case *CmdOnlySegment:
2016-11-27 15:39:09 -05:00
v.HandleOption(seg.Option)
2016-12-08 10:27:41 -05:00
if seg.Command() == CommandTerminate {
2016-11-27 15:39:09 -05:00
state := v.State()
2016-07-05 17:53:46 -04:00
if state == StateActive ||
state == StatePeerClosed {
2016-11-27 15:39:09 -05:00
v.SetState(StatePeerTerminating)
} else if state == StateReadyToClose {
2016-11-27 15:39:09 -05:00
v.SetState(StateTerminating)
2016-07-05 17:53:46 -04:00
} else if state == StateTerminating {
2016-11-27 15:39:09 -05:00
v.SetState(StateTerminated)
2016-07-05 17:02:52 -04:00
}
}
2016-12-26 02:22:25 -05:00
if seg.Option == SegmentOptionClose || seg.Command() == CommandTerminate {
v.OnDataInput()
v.OnDataOutput()
}
2016-11-27 15:39:09 -05:00
v.sendingWorker.ProcessReceivingNext(seg.ReceivinNext)
v.receivingWorker.ProcessSendingNext(seg.SendingNext)
v.roundTrip.UpdatePeerRTO(seg.PeerRTO, current)
2016-07-13 17:15:40 -04:00
seg.Release()
2016-07-05 17:02:52 -04:00
default:
}
}
}
2016-11-27 15:39:09 -05:00
func (v *Connection) flush() {
current := v.Elapsed()
2016-07-05 17:02:52 -04:00
2016-11-27 15:39:09 -05:00
if v.State() == StateTerminated {
2016-07-05 17:02:52 -04:00
return
}
2016-11-27 15:39:09 -05:00
if v.State() == StateActive && current-atomic.LoadUint32(&v.lastIncomingTime) >= 30000 {
v.Close()
2016-07-05 17:02:52 -04:00
}
2016-11-27 15:39:09 -05:00
if v.State() == StateReadyToClose && v.sendingWorker.IsEmpty() {
v.SetState(StateTerminating)
2016-07-12 17:54:54 -04:00
}
2016-07-05 17:02:52 -04:00
2016-11-27 15:39:09 -05:00
if v.State() == StateTerminating {
log.Debug("KCP|Connection: #", v.conv, " sending terminating cmd.")
v.Ping(current, CommandTerminate)
2016-07-05 17:02:52 -04:00
2016-11-27 15:39:09 -05:00
if current-atomic.LoadUint32(&v.stateBeginTime) > 8000 {
v.SetState(StateTerminated)
2016-07-05 17:02:52 -04:00
}
return
}
2016-11-27 15:39:09 -05:00
if v.State() == StatePeerTerminating && current-atomic.LoadUint32(&v.stateBeginTime) > 4000 {
v.SetState(StateTerminating)
}
2016-07-05 17:02:52 -04:00
2016-11-27 15:39:09 -05:00
if v.State() == StateReadyToClose && current-atomic.LoadUint32(&v.stateBeginTime) > 15000 {
v.SetState(StateTerminating)
2016-07-05 17:02:52 -04:00
}
// flush acknowledges
2016-11-27 15:39:09 -05:00
v.receivingWorker.Flush(current)
v.sendingWorker.Flush(current)
2016-07-05 17:02:52 -04:00
2016-11-27 15:39:09 -05:00
if current-atomic.LoadUint32(&v.lastPingTime) >= 3000 {
v.Ping(current, CommandPing)
2016-07-05 17:02:52 -04:00
}
}
2016-11-27 15:39:09 -05:00
func (v *Connection) State() State {
return State(atomic.LoadInt32((*int32)(&v.state)))
2016-07-05 17:02:52 -04:00
}
2016-10-11 06:24:19 -04:00
2016-11-27 15:39:09 -05:00
func (v *Connection) Ping(current uint32, cmd Command) {
2016-10-11 06:24:19 -04:00
seg := NewCmdOnlySegment()
2016-11-27 15:39:09 -05:00
seg.Conv = v.conv
2016-12-08 10:27:41 -05:00
seg.Cmd = cmd
2016-11-27 15:39:09 -05:00
seg.ReceivinNext = v.receivingWorker.nextNumber
seg.SendingNext = v.sendingWorker.firstUnacknowledged
seg.PeerRTO = v.roundTrip.Timeout()
if v.State() == StateReadyToClose {
2016-10-11 06:24:19 -04:00
seg.Option = SegmentOptionClose
}
2016-11-27 15:39:09 -05:00
v.output.Write(seg)
atomic.StoreUint32(&v.lastPingTime, current)
2016-10-11 06:24:19 -04:00
seg.Release()
}