forked from TrueCloudLab/neoneo-go
network/config: redesign ping timeout handling a bit
1) Make timeout a timeout, don't do magic ping counts. 2) Drop additional timer from the main peer's protocol loop, create it dynamically and make it disconnect the peer. 3) Don't expose the ping counter to the outside, handle more logic inside the Peer. Relates to #430.
This commit is contained in:
parent
62092c703d
commit
2c4ace022e
13 changed files with 67 additions and 61 deletions
|
@ -50,8 +50,8 @@ ApplicationConfiguration:
|
||||||
Relay: true
|
Relay: true
|
||||||
DialTimeout: 3
|
DialTimeout: 3
|
||||||
ProtoTickInterval: 2
|
ProtoTickInterval: 2
|
||||||
PingInterval: 60
|
PingInterval: 30
|
||||||
PingTimeout: 60
|
PingTimeout: 90
|
||||||
MaxPeers: 100
|
MaxPeers: 100
|
||||||
AttemptConnPeers: 20
|
AttemptConnPeers: 20
|
||||||
MinPeers: 5
|
MinPeers: 5
|
||||||
|
|
|
@ -41,8 +41,8 @@ ApplicationConfiguration:
|
||||||
Relay: true
|
Relay: true
|
||||||
DialTimeout: 3
|
DialTimeout: 3
|
||||||
ProtoTickInterval: 2
|
ProtoTickInterval: 2
|
||||||
PingInterval: 60
|
PingInterval: 30
|
||||||
PingTimeout: 60
|
PingTimeout: 90
|
||||||
MaxPeers: 10
|
MaxPeers: 10
|
||||||
AttemptConnPeers: 5
|
AttemptConnPeers: 5
|
||||||
MinPeers: 3
|
MinPeers: 3
|
||||||
|
|
|
@ -41,8 +41,8 @@ ApplicationConfiguration:
|
||||||
Relay: true
|
Relay: true
|
||||||
DialTimeout: 3
|
DialTimeout: 3
|
||||||
ProtoTickInterval: 2
|
ProtoTickInterval: 2
|
||||||
PingInterval: 60
|
PingInterval: 30
|
||||||
PingTimeout: 60
|
PingTimeout: 90
|
||||||
MaxPeers: 10
|
MaxPeers: 10
|
||||||
AttemptConnPeers: 5
|
AttemptConnPeers: 5
|
||||||
MinPeers: 3
|
MinPeers: 3
|
||||||
|
|
|
@ -35,8 +35,8 @@ ApplicationConfiguration:
|
||||||
Relay: true
|
Relay: true
|
||||||
DialTimeout: 3
|
DialTimeout: 3
|
||||||
ProtoTickInterval: 2
|
ProtoTickInterval: 2
|
||||||
PingInterval: 60
|
PingInterval: 30
|
||||||
PingTimeout: 60
|
PingTimeout: 90
|
||||||
MaxPeers: 10
|
MaxPeers: 10
|
||||||
AttemptConnPeers: 5
|
AttemptConnPeers: 5
|
||||||
MinPeers: 0
|
MinPeers: 0
|
||||||
|
|
|
@ -41,8 +41,8 @@ ApplicationConfiguration:
|
||||||
Relay: true
|
Relay: true
|
||||||
DialTimeout: 3
|
DialTimeout: 3
|
||||||
ProtoTickInterval: 2
|
ProtoTickInterval: 2
|
||||||
PingInterval: 60
|
PingInterval: 30
|
||||||
PingTimeout: 60
|
PingTimeout: 90
|
||||||
MaxPeers: 10
|
MaxPeers: 10
|
||||||
AttemptConnPeers: 5
|
AttemptConnPeers: 5
|
||||||
MinPeers: 3
|
MinPeers: 3
|
||||||
|
|
|
@ -41,8 +41,8 @@ ApplicationConfiguration:
|
||||||
Relay: true
|
Relay: true
|
||||||
DialTimeout: 3
|
DialTimeout: 3
|
||||||
ProtoTickInterval: 2
|
ProtoTickInterval: 2
|
||||||
PingInterval: 60
|
PingInterval: 30
|
||||||
PingTimeout: 60
|
PingTimeout: 90
|
||||||
MaxPeers: 10
|
MaxPeers: 10
|
||||||
AttemptConnPeers: 5
|
AttemptConnPeers: 5
|
||||||
MinPeers: 3
|
MinPeers: 3
|
||||||
|
|
|
@ -41,8 +41,8 @@ ApplicationConfiguration:
|
||||||
Relay: true
|
Relay: true
|
||||||
DialTimeout: 3
|
DialTimeout: 3
|
||||||
ProtoTickInterval: 2
|
ProtoTickInterval: 2
|
||||||
PingInterval: 60
|
PingInterval: 30
|
||||||
PingTimeout: 60
|
PingTimeout: 90
|
||||||
MaxPeers: 10
|
MaxPeers: 10
|
||||||
AttemptConnPeers: 5
|
AttemptConnPeers: 5
|
||||||
MinPeers: 3
|
MinPeers: 3
|
||||||
|
|
|
@ -50,8 +50,8 @@ ApplicationConfiguration:
|
||||||
Relay: true
|
Relay: true
|
||||||
DialTimeout: 3
|
DialTimeout: 3
|
||||||
ProtoTickInterval: 2
|
ProtoTickInterval: 2
|
||||||
PingInterval: 60
|
PingInterval: 30
|
||||||
PingTimeout: 60
|
PingTimeout: 90
|
||||||
MaxPeers: 100
|
MaxPeers: 100
|
||||||
AttemptConnPeers: 20
|
AttemptConnPeers: 20
|
||||||
MinPeers: 5
|
MinPeers: 5
|
||||||
|
|
|
@ -40,8 +40,8 @@ ApplicationConfiguration:
|
||||||
Relay: true
|
Relay: true
|
||||||
DialTimeout: 3
|
DialTimeout: 3
|
||||||
ProtoTickInterval: 2
|
ProtoTickInterval: 2
|
||||||
PingInterval: 60
|
PingInterval: 30
|
||||||
PingTimeout: 60
|
PingTimeout: 90
|
||||||
MaxPeers: 50
|
MaxPeers: 50
|
||||||
AttemptConnPeers: 5
|
AttemptConnPeers: 5
|
||||||
MinPeers: 1
|
MinPeers: 1
|
||||||
|
|
|
@ -160,7 +160,7 @@ type localPeer struct {
|
||||||
handshaked bool
|
handshaked bool
|
||||||
t *testing.T
|
t *testing.T
|
||||||
messageHandler func(t *testing.T, msg *Message)
|
messageHandler func(t *testing.T, msg *Message)
|
||||||
pingSent int
|
pingSent int
|
||||||
}
|
}
|
||||||
|
|
||||||
func newLocalPeer(t *testing.T) *localPeer {
|
func newLocalPeer(t *testing.T) *localPeer {
|
||||||
|
@ -206,9 +206,6 @@ func (p *localPeer) Version() *payload.Version {
|
||||||
func (p *localPeer) LastBlockIndex() uint32 {
|
func (p *localPeer) LastBlockIndex() uint32 {
|
||||||
return p.lastBlockIndex
|
return p.lastBlockIndex
|
||||||
}
|
}
|
||||||
func (p *localPeer) UpdateLastBlockIndex(newIndex uint32) {
|
|
||||||
p.lastBlockIndex = newIndex
|
|
||||||
}
|
|
||||||
func (p *localPeer) HandleVersion(v *payload.Version) error {
|
func (p *localPeer) HandleVersion(v *payload.Version) error {
|
||||||
p.version = v
|
p.version = v
|
||||||
return nil
|
return nil
|
||||||
|
@ -225,11 +222,14 @@ func (p *localPeer) HandleVersionAck() error {
|
||||||
p.handshaked = true
|
p.handshaked = true
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
func (p *localPeer) GetPingSent() int {
|
func (p *localPeer) SendPing() error {
|
||||||
return p.pingSent
|
p.pingSent++
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
func (p *localPeer) UpdatePingSent(newValue int) {
|
func (p *localPeer) HandlePong(pong *payload.Ping) error {
|
||||||
p.pingSent = newValue
|
p.lastBlockIndex = pong.LastBlockIndex
|
||||||
|
p.pingSent--
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *localPeer) Handshaked() bool {
|
func (p *localPeer) Handshaked() bool {
|
||||||
|
|
|
@ -36,8 +36,12 @@ type Peer interface {
|
||||||
EnqueueHPPacket([]byte) error
|
EnqueueHPPacket([]byte) error
|
||||||
Version() *payload.Version
|
Version() *payload.Version
|
||||||
LastBlockIndex() uint32
|
LastBlockIndex() uint32
|
||||||
UpdateLastBlockIndex(lbIndex uint32)
|
|
||||||
Handshaked() bool
|
Handshaked() bool
|
||||||
|
|
||||||
|
// SendPing enqueues a ping message to be sent to the peer and does
|
||||||
|
// appropriate protocol handling like timeouts and outstanding pings
|
||||||
|
// management.
|
||||||
|
SendPing() error
|
||||||
SendVersion(*Message) error
|
SendVersion(*Message) error
|
||||||
SendVersionAck(*Message) error
|
SendVersionAck(*Message) error
|
||||||
// StartProtocol is a goroutine to be run after the handshake. It
|
// StartProtocol is a goroutine to be run after the handshake. It
|
||||||
|
@ -45,6 +49,7 @@ type Peer interface {
|
||||||
StartProtocol()
|
StartProtocol()
|
||||||
HandleVersion(*payload.Version) error
|
HandleVersion(*payload.Version) error
|
||||||
HandleVersionAck() error
|
HandleVersionAck() error
|
||||||
GetPingSent() int
|
|
||||||
UpdatePingSent(int)
|
// HandlePong checks pong contents against Peer's state and updates it.
|
||||||
|
HandlePong(pong *payload.Ping) error
|
||||||
}
|
}
|
||||||
|
|
|
@ -29,7 +29,6 @@ const (
|
||||||
maxBlockBatch = 200
|
maxBlockBatch = 200
|
||||||
maxAddrsToSend = 200
|
maxAddrsToSend = 200
|
||||||
minPoolCount = 30
|
minPoolCount = 30
|
||||||
defaultPingLimit = 4
|
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
|
@ -373,12 +372,10 @@ func (s *Server) handlePing(p Peer, ping *payload.Ping) error {
|
||||||
|
|
||||||
// handlePing processes pong request.
|
// handlePing processes pong request.
|
||||||
func (s *Server) handlePong(p Peer, pong *payload.Ping) error {
|
func (s *Server) handlePong(p Peer, pong *payload.Ping) error {
|
||||||
pingSent := p.GetPingSent()
|
err := p.HandlePong(pong)
|
||||||
if pingSent == 0 {
|
if err != nil {
|
||||||
return errors.New("pong message wasn't expected")
|
return err
|
||||||
}
|
}
|
||||||
p.UpdatePingSent(pingSent - 1)
|
|
||||||
p.UpdateLastBlockIndex(pong.LastBlockIndex)
|
|
||||||
if s.chain.HeaderHeight() < pong.LastBlockIndex {
|
if s.chain.HeaderHeight() < pong.LastBlockIndex {
|
||||||
return s.requestHeaders(p)
|
return s.requestHeaders(p)
|
||||||
}
|
}
|
||||||
|
|
|
@ -25,7 +25,9 @@ const (
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
errStateMismatch = errors.New("tried to send protocol message before handshake completed")
|
errStateMismatch = errors.New("tried to send protocol message before handshake completed")
|
||||||
|
errPingPong = errors.New("ping/pong timeout")
|
||||||
|
errUnexpectedPong = errors.New("pong message wasn't expected")
|
||||||
)
|
)
|
||||||
|
|
||||||
// TCPPeer represents a connected remote node in the
|
// TCPPeer represents a connected remote node in the
|
||||||
|
@ -51,7 +53,8 @@ type TCPPeer struct {
|
||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
|
|
||||||
// number of sent pings.
|
// number of sent pings.
|
||||||
pingSent int
|
pingSent int
|
||||||
|
pingTimer *time.Timer
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewTCPPeer returns a TCPPeer structure based on the given connection.
|
// NewTCPPeer returns a TCPPeer structure based on the given connection.
|
||||||
|
@ -191,7 +194,6 @@ func (p *TCPPeer) StartProtocol() {
|
||||||
}
|
}
|
||||||
|
|
||||||
timer := time.NewTimer(p.server.ProtoTickInterval)
|
timer := time.NewTimer(p.server.ProtoTickInterval)
|
||||||
pingTimer := time.NewTimer(p.server.PingTimeout)
|
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-p.done:
|
case <-p.done:
|
||||||
|
@ -210,20 +212,12 @@ func (p *TCPPeer) StartProtocol() {
|
||||||
} else {
|
} else {
|
||||||
diff := time.Now().UTC().Unix() - p.server.getLastBlockTime()
|
diff := time.Now().UTC().Unix() - p.server.getLastBlockTime()
|
||||||
if diff > int64(p.server.PingInterval/time.Second) {
|
if diff > int64(p.server.PingInterval/time.Second) {
|
||||||
p.UpdatePingSent(p.GetPingSent() + 1)
|
err = p.SendPing()
|
||||||
err = p.EnqueueMessage(NewMessage(p.server.Net, CMDPing, payload.NewPing(p.server.id, p.server.chain.HeaderHeight())))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err == nil {
|
if err == nil {
|
||||||
timer.Reset(p.server.ProtoTickInterval)
|
timer.Reset(p.server.ProtoTickInterval)
|
||||||
}
|
}
|
||||||
case <-pingTimer.C:
|
|
||||||
if p.GetPingSent() > defaultPingLimit {
|
|
||||||
err = errors.New("ping/pong timeout")
|
|
||||||
} else {
|
|
||||||
pingTimer.Reset(p.server.PingTimeout)
|
|
||||||
p.UpdatePingSent(0)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
timer.Stop()
|
timer.Stop()
|
||||||
|
@ -350,23 +344,33 @@ func (p *TCPPeer) LastBlockIndex() uint32 {
|
||||||
return p.lastBlockIndex
|
return p.lastBlockIndex
|
||||||
}
|
}
|
||||||
|
|
||||||
// UpdateLastBlockIndex updates last block index.
|
// SendPing sends a ping message to the peer and does appropriate accounting of
|
||||||
func (p *TCPPeer) UpdateLastBlockIndex(newIndex uint32) {
|
// outstanding pings and timeouts.
|
||||||
|
func (p *TCPPeer) SendPing() error {
|
||||||
p.lock.Lock()
|
p.lock.Lock()
|
||||||
defer p.lock.Unlock()
|
p.pingSent++
|
||||||
p.lastBlockIndex = newIndex
|
if p.pingTimer == nil {
|
||||||
|
p.pingTimer = time.AfterFunc(p.server.PingTimeout, func() {
|
||||||
|
p.Disconnect(errPingPong)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
p.lock.Unlock()
|
||||||
|
return p.EnqueueMessage(NewMessage(p.server.Net, CMDPing, payload.NewPing(p.server.id, p.server.chain.HeaderHeight())))
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetPingSent returns flag whether ping was sent or not.
|
// HandlePong handles a pong message received from the peer and does appropriate
|
||||||
func (p *TCPPeer) GetPingSent() int {
|
// accounting of outstanding pings and timeouts.
|
||||||
p.lock.RLock()
|
func (p *TCPPeer) HandlePong(pong *payload.Ping) error {
|
||||||
defer p.lock.RUnlock()
|
|
||||||
return p.pingSent
|
|
||||||
}
|
|
||||||
|
|
||||||
// UpdatePingSent updates pingSent value.
|
|
||||||
func (p *TCPPeer) UpdatePingSent(newValue int) {
|
|
||||||
p.lock.Lock()
|
p.lock.Lock()
|
||||||
defer p.lock.Unlock()
|
defer p.lock.Unlock()
|
||||||
p.pingSent = newValue
|
if p.pingTimer != nil && !p.pingTimer.Stop() {
|
||||||
|
return errPingPong
|
||||||
|
}
|
||||||
|
p.pingTimer = nil
|
||||||
|
p.pingSent--
|
||||||
|
if p.pingSent < 0 {
|
||||||
|
return errUnexpectedPong
|
||||||
|
}
|
||||||
|
p.lastBlockIndex = pong.LastBlockIndex
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue