Merge pull request #1675 from cfromknecht/peer-backoff-backoff
server: peer backoff backoff
This commit is contained in:
2 files changed
+48
-11
No files matched your search
@@ -111,13 +111,12 @@ type peer struct {
|
|||||||
addr *lnwire.NetAddress
|
addr *lnwire.NetAddress
|
||||||
pubKeyBytes [33]byte
|
pubKeyBytes [33]byte
|
||||||
|
|
||||||
inbound bool
|
// startTime is the time this peer connection was successfully
|
||||||
|
// established. It will be zero for peers that did not successfuly
|
||||||
|
// Start().
|
||||||
|
startTime time.Time
|
||||||
|
|
||||||
// This mutex protects all the stats below it.
|
inbound bool
|
||||||
sync.RWMutex
|
|
||||||
timeConnected time.Time
|
|
||||||
lastSend time.Time
|
|
||||||
lastRecv time.Time
|
|
||||||
|
|
||||||
// sendQueue is the channel which is used to queue outgoing to be
|
// sendQueue is the channel which is used to queue outgoing to be
|
||||||
// written onto the wire. Note that this channel is unbuffered.
|
// written onto the wire. Note that this channel is unbuffered.
|
||||||
@@ -303,6 +302,8 @@ func (p *peer) Start() error {
|
|||||||
return fmt.Errorf("unable to load channels: %v", err)
|
return fmt.Errorf("unable to load channels: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
p.startTime = time.Now()
|
||||||
|
|
||||||
p.wg.Add(5)
|
p.wg.Add(5)
|
||||||
go p.queueHandler()
|
go p.queueHandler()
|
||||||
go p.writeHandler()
|
go p.writeHandler()
|
||||||
@@ -1977,6 +1978,8 @@ func (p *peer) sendInitMsg() error {
|
|||||||
// SendMessage sends a variadic number of message to remote peer. The first
|
// SendMessage sends a variadic number of message to remote peer. The first
|
||||||
// argument denotes if the method should block until the message has been sent
|
// argument denotes if the method should block until the message has been sent
|
||||||
// to the remote peer.
|
// to the remote peer.
|
||||||
|
//
|
||||||
|
// NOTE: Part of the lnpeer.Peer interface.
|
||||||
func (p *peer) SendMessage(sync bool, msgs ...lnwire.Message) error {
|
func (p *peer) SendMessage(sync bool, msgs ...lnwire.Message) error {
|
||||||
// Add all incoming messages to the outgoing queue. A list of error
|
// Add all incoming messages to the outgoing queue. A list of error
|
||||||
// chans is populated for each message if the caller requested a sync
|
// chans is populated for each message if the caller requested a sync
|
||||||
@@ -2009,22 +2012,30 @@ func (p *peer) SendMessage(sync bool, msgs ...lnwire.Message) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// PubKey returns the pubkey of the peer in compressed serialized format.
|
// PubKey returns the pubkey of the peer in compressed serialized format.
|
||||||
|
//
|
||||||
|
// NOTE: Part of the lnpeer.Peer interface.
|
||||||
func (p *peer) PubKey() [33]byte {
|
func (p *peer) PubKey() [33]byte {
|
||||||
return p.pubKeyBytes
|
return p.pubKeyBytes
|
||||||
}
|
}
|
||||||
|
|
||||||
// IdentityKey returns the public key of the remote peer.
|
// IdentityKey returns the public key of the remote peer.
|
||||||
|
//
|
||||||
|
// NOTE: Part of the lnpeer.Peer interface.
|
||||||
func (p *peer) IdentityKey() *btcec.PublicKey {
|
func (p *peer) IdentityKey() *btcec.PublicKey {
|
||||||
return p.addr.IdentityKey
|
return p.addr.IdentityKey
|
||||||
}
|
}
|
||||||
|
|
||||||
// Address returns the network address of the remote peer.
|
// Address returns the network address of the remote peer.
|
||||||
|
//
|
||||||
|
// NOTE: Part of the lnpeer.Peer interface.
|
||||||
func (p *peer) Address() net.Addr {
|
func (p *peer) Address() net.Addr {
|
||||||
return p.addr.Address
|
return p.addr.Address
|
||||||
}
|
}
|
||||||
|
|
||||||
// AddNewChannel adds a new channel to the peer. The channel should fail to be
|
// AddNewChannel adds a new channel to the peer. The channel should fail to be
|
||||||
// added if the cancel channel is closed.
|
// added if the cancel channel is closed.
|
||||||
|
//
|
||||||
|
// NOTE: Part of the lnpeer.Peer interface.
|
||||||
func (p *peer) AddNewChannel(channel *lnwallet.LightningChannel,
|
func (p *peer) AddNewChannel(channel *lnwallet.LightningChannel,
|
||||||
cancel <-chan struct{}) error {
|
cancel <-chan struct{}) error {
|
||||||
|
|
||||||
@@ -2053,6 +2064,12 @@ func (p *peer) AddNewChannel(channel *lnwallet.LightningChannel,
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// StartTime returns the time at which the connection was established if the
|
||||||
|
// peer started successfully, and zero otherwise.
|
||||||
|
func (p *peer) StartTime() time.Time {
|
||||||
|
return p.startTime
|
||||||
|
}
|
||||||
|
|
||||||
// TODO(roasbeef): make all start/stop mutexes a CAS
|
// TODO(roasbeef): make all start/stop mutexes a CAS
|
||||||
|
|
||||||
// fetchLastChanUpdate returns a function which is able to retrieve the last
|
// fetchLastChanUpdate returns a function which is able to retrieve the last
|
||||||
|
|||||||
@@ -2018,7 +2018,7 @@ func (s *server) peerTerminationWatcher(p *peer) {
|
|||||||
s.persistentConnReqs[pubStr], connReq)
|
s.persistentConnReqs[pubStr], connReq)
|
||||||
|
|
||||||
// Record the computed backoff in the backoff map.
|
// Record the computed backoff in the backoff map.
|
||||||
backoff := s.nextPeerBackoff(pubStr)
|
backoff := s.nextPeerBackoff(pubStr, p.StartTime())
|
||||||
s.persistentPeersBackoff[pubStr] = backoff
|
s.persistentPeersBackoff[pubStr] = backoff
|
||||||
|
|
||||||
// Initialize a retry canceller for this peer if one does not
|
// Initialize a retry canceller for this peer if one does not
|
||||||
@@ -2055,7 +2055,9 @@ func (s *server) peerTerminationWatcher(p *peer) {
|
|||||||
// nextPeerBackoff computes the next backoff duration for a peer's pubkey using
|
// nextPeerBackoff computes the next backoff duration for a peer's pubkey using
|
||||||
// exponential backoff. If no previous backoff was known, the default is
|
// exponential backoff. If no previous backoff was known, the default is
|
||||||
// returned.
|
// returned.
|
||||||
func (s *server) nextPeerBackoff(pubStr string) time.Duration {
|
func (s *server) nextPeerBackoff(pubStr string,
|
||||||
|
startTime time.Time) time.Duration {
|
||||||
|
|
||||||
// Now, determine the appropriate backoff to use for the retry.
|
// Now, determine the appropriate backoff to use for the retry.
|
||||||
backoff, ok := s.persistentPeersBackoff[pubStr]
|
backoff, ok := s.persistentPeersBackoff[pubStr]
|
||||||
if !ok {
|
if !ok {
|
||||||
@@ -2063,9 +2065,27 @@ func (s *server) nextPeerBackoff(pubStr string) time.Duration {
|
|||||||
return defaultBackoff
|
return defaultBackoff
|
||||||
}
|
}
|
||||||
|
|
||||||
// Otherwise, use a previous backoff to compute the
|
// If the peer failed to start properly, we'll just use the previous
|
||||||
// subsequent randomized exponential backoff duration.
|
// backoff to compute the subsequent randomized exponential backoff
|
||||||
return computeNextBackoff(backoff)
|
// duration. This will roughly double on average.
|
||||||
|
if startTime.IsZero() {
|
||||||
|
return computeNextBackoff(backoff)
|
||||||
|
}
|
||||||
|
|
||||||
|
// The peer succeeded in starting. We'll reduce the timeout duration
|
||||||
|
// by the length of the connection before applying randomized
|
||||||
|
// exponential backoff. We'll only apply this if:
|
||||||
|
// backoff - connDuration > defaultBackoff
|
||||||
|
connDuration := time.Now().Sub(startTime)
|
||||||
|
relaxedBackoff := backoff - connDuration
|
||||||
|
if relaxedBackoff > defaultBackoff {
|
||||||
|
return computeNextBackoff(relaxedBackoff)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Otherwise, backoff - connDuration <= defaultBackoff, meaning the
|
||||||
|
// connection lasted much longer than our previous backoff. To reward
|
||||||
|
// such good behavior, we'll reconnect after the default timeout.
|
||||||
|
return defaultBackoff
|
||||||
}
|
}
|
||||||
|
|
||||||
// shouldRequestGraphSync returns true if the servers deems it necessary that
|
// shouldRequestGraphSync returns true if the servers deems it necessary that
|
||||||
|
|||||||
Reference in new issue
Block a user