Merge pull request #610 from nspcc-dev/ping-improvements

Ping improvements

config/protocol.mainnet.yml

@@ -50,8 +50,8 @@ ApplicationConfiguration:
   Relay: true
   DialTimeout: 3
   ProtoTickInterval: 2
-  PingInterval: 60
+  PingInterval: 30
-  PingTimeout: 60
+  PingTimeout: 90
   MaxPeers: 100
   AttemptConnPeers: 20
   MinPeers: 5

config/protocol.privnet.docker.four.yml

@@ -41,8 +41,8 @@ ApplicationConfiguration:
   Relay: true
   DialTimeout: 3
   ProtoTickInterval: 2
-  PingInterval: 60
+  PingInterval: 30
-  PingTimeout: 60
+  PingTimeout: 90
   MaxPeers: 10
   AttemptConnPeers: 5
   MinPeers: 3

config/protocol.privnet.docker.one.yml

@@ -41,8 +41,8 @@ ApplicationConfiguration:
   Relay: true
   DialTimeout: 3
   ProtoTickInterval: 2
-  PingInterval: 60
+  PingInterval: 30
-  PingTimeout: 60
+  PingTimeout: 90
   MaxPeers: 10
   AttemptConnPeers: 5
   MinPeers: 3

config/protocol.privnet.docker.single.yml

@@ -35,8 +35,8 @@ ApplicationConfiguration:
   Relay: true
   DialTimeout: 3
   ProtoTickInterval: 2
-  PingInterval: 60
+  PingInterval: 30
-  PingTimeout: 60
+  PingTimeout: 90
   MaxPeers: 10
   AttemptConnPeers: 5
   MinPeers: 0

config/protocol.privnet.docker.three.yml

@@ -41,8 +41,8 @@ ApplicationConfiguration:
   Relay: true
   DialTimeout: 3
   ProtoTickInterval: 2
-  PingInterval: 60
+  PingInterval: 30
-  PingTimeout: 60
+  PingTimeout: 90
   MaxPeers: 10
   AttemptConnPeers: 5
   MinPeers: 3

config/protocol.privnet.docker.two.yml

@@ -41,8 +41,8 @@ ApplicationConfiguration:
   Relay: true
   DialTimeout: 3
   ProtoTickInterval: 2
-  PingInterval: 60
+  PingInterval: 30
-  PingTimeout: 60
+  PingTimeout: 90
   MaxPeers: 10
   AttemptConnPeers: 5
   MinPeers: 3

config/protocol.privnet.yml

@@ -41,8 +41,8 @@ ApplicationConfiguration:
   Relay: true
   DialTimeout: 3
   ProtoTickInterval: 2
-  PingInterval: 60
+  PingInterval: 30
-  PingTimeout: 60
+  PingTimeout: 90
   MaxPeers: 10
   AttemptConnPeers: 5
   MinPeers: 3

config/protocol.testnet.yml

@@ -50,8 +50,8 @@ ApplicationConfiguration:
   Relay: true
   DialTimeout: 3
   ProtoTickInterval: 2
-  PingInterval: 60
+  PingInterval: 30
-  PingTimeout: 60
+  PingTimeout: 90
   MaxPeers: 100
   AttemptConnPeers: 20
   MinPeers: 5

config/protocol.unit_testnet.yml

@@ -40,8 +40,8 @@ ApplicationConfiguration:
   Relay: true
   DialTimeout: 3
   ProtoTickInterval: 2
-  PingInterval: 60
+  PingInterval: 30
-  PingTimeout: 60
+  PingTimeout: 90
   MaxPeers: 50
   AttemptConnPeers: 5
   MinPeers: 1

pkg/network/helper_test.go

@@ -206,9 +206,6 @@ func (p *localPeer) Version() *payload.Version {
 func (p *localPeer) LastBlockIndex() uint32 {
 	return p.lastBlockIndex
 }
-func (p *localPeer) UpdateLastBlockIndex(newIndex uint32) {
-	p.lastBlockIndex = newIndex
-}
 func (p *localPeer) HandleVersion(v *payload.Version) error {
 	p.version = v
 	return nil
@@ -225,11 +222,14 @@ func (p *localPeer) HandleVersionAck() error {
 	p.handshaked = true
 	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 {

pkg/network/peer.go

@@ -36,8 +36,12 @@ type Peer interface {
 	EnqueueHPPacket([]byte) error
 	Version() *payload.Version
 	LastBlockIndex() uint32
-	UpdateLastBlockIndex(lbIndex uint32)
 	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
 	SendVersionAck(*Message) error
 	// StartProtocol is a goroutine to be run after the handshake. It
@@ -45,6 +49,7 @@ type Peer interface {
 	StartProtocol()
 	HandleVersion(*payload.Version) error
 	HandleVersionAck() error
-	GetPingSent() int
+
-	UpdatePingSent(int)
+	// HandlePong checks pong contents against Peer's state and updates it.
+	HandlePong(pong *payload.Ping) error
 }

pkg/network/server.go

@@ -29,7 +29,6 @@ const (
 	maxBlockBatch           = 200
 	maxAddrsToSend          = 200
 	minPoolCount            = 30
-	defaultPingLimit        = 4
 )
 
 var (
@@ -68,6 +67,8 @@ type (
 		quit       chan struct{}
 
 		connected *atomic.Bool
+		// Time of the last block receival.
+		lastBlockTS *atomic.Int64
 
 		log *zap.Logger
 	}
@@ -101,6 +102,7 @@ func NewServer(config ServerConfig, chain core.Blockchainer, log *zap.Logger) *S
 		unregister:   make(chan peerDrop),
 		peers:        make(map[Peer]bool),
 		connected:    atomic.NewBool(false),
+		lastBlockTS:  atomic.NewInt64(0),
 		log:          log,
 	}
 
@@ -359,6 +361,7 @@ func (s *Server) handleHeadersCmd(p Peer, headers *payload.Headers) {
 
 // handleBlockCmd processes the received block received from its peer.
 func (s *Server) handleBlockCmd(p Peer, block *block.Block) error {
+	s.lastBlockTS.Store(time.Now().UTC().Unix())
 	return s.bQueue.putBlock(block)
 }
 
@@ -369,12 +372,10 @@ func (s *Server) handlePing(p Peer, ping *payload.Ping) error {
 
 // handlePing processes pong request.
 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 {
 		return s.requestHeaders(p)
 	}
@@ -574,7 +575,7 @@ func (s *Server) requestBlocks(p Peer) error {
 	if len(hashes) > 0 {
 		payload := payload.NewInventory(payload.BlockType, hashes)
 		return p.EnqueueMessage(NewMessage(s.Net, CMDGetData, payload))
-	} else if s.chain.HeaderHeight() < p.Version().StartHeight {
+	} else if s.chain.HeaderHeight() < p.LastBlockIndex() {
 		return s.requestHeaders(p)
 	}
 	return nil
@@ -658,6 +659,12 @@ func (s *Server) handleNewPayload(p *consensus.Payload) {
 	s.relayInventoryCmd(CMDInv, payload.ConsensusType, p.Hash())
 }
 
+// getLastBlockTime returns unix timestamp for the moment when the last block
+// was received.
+func (s *Server) getLastBlockTime() int64 {
+	return s.lastBlockTS.Load()
+}
+
 func (s *Server) requestTx(hashes ...util.Uint256) {
 	if len(hashes) == 0 {
 		return

pkg/network/tcp_peer.go

@@ -26,6 +26,8 @@ const (
 
 var (
 	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
@@ -52,6 +54,7 @@ type TCPPeer struct {
 
 	// number of sent pings.
 	pingSent  int
+	pingTimer *time.Timer
 }
 
 // NewTCPPeer returns a TCPPeer structure based on the given connection.
@@ -191,7 +194,6 @@ func (p *TCPPeer) StartProtocol() {
 	}
 
 	timer := time.NewTimer(p.server.ProtoTickInterval)
-	pingTimer := time.NewTimer(p.server.PingTimeout)
 	for {
 		select {
 		case <-p.done:
@@ -207,29 +209,15 @@ func (p *TCPPeer) StartProtocol() {
 			// Try to sync in headers and block with the peer if his block height is higher then ours.
 			if p.LastBlockIndex() > p.server.chain.BlockHeight() {
 				err = p.server.requestBlocks(p)
+			} else {
+				diff := time.Now().UTC().Unix() - p.server.getLastBlockTime()
+				if diff > int64(p.server.PingInterval/time.Second) {
+					err = p.SendPing()
+				}
 			}
 			if err == nil {
 				timer.Reset(p.server.ProtoTickInterval)
 			}
-			if p.server.chain.HeaderHeight() >= p.LastBlockIndex() {
-				block, errGetBlock := p.server.chain.GetBlock(p.server.chain.CurrentBlockHash())
-				if errGetBlock != nil {
-					err = errGetBlock
-				} else {
-					diff := uint32(time.Now().UTC().Unix()) - block.Timestamp
-					if diff > uint32(p.server.PingInterval/time.Second) {
-						p.UpdatePingSent(p.GetPingSent() + 1)
-						err = p.EnqueueMessage(NewMessage(p.server.Net, CMDPing, payload.NewPing(p.server.id, p.server.chain.HeaderHeight())))
-					}
-				}
-			}
-		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 {
 			timer.Stop()
@@ -356,23 +344,33 @@ func (p *TCPPeer) LastBlockIndex() uint32 {
 	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()
-	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()
 	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
 }