network: implement adaptive peer requests

When the network is big enough, MinPeers may be suboptimal for good network
connectivity, but if we know the network size we can do some estimation on the
number of sufficient peers.
This commit is contained in:
Roman Khimov 2022-10-13 22:53:20 +03:00
parent c17b2afab5
commit 851cbc7dab
3 changed files with 33 additions and 2 deletions

View file

@ -19,6 +19,7 @@ const (
type Discoverer interface { type Discoverer interface {
BackFill(...string) BackFill(...string)
GetFanOut() int GetFanOut() int
NetworkSize() int
PoolCount() int PoolCount() int
RequestRemote(int) RequestRemote(int)
RegisterBadAddr(string) RegisterBadAddr(string)
@ -48,6 +49,7 @@ type DefaultDiscovery struct {
unconnectedAddrs map[string]int unconnectedAddrs map[string]int
attempted map[string]bool attempted map[string]bool
optimalFanOut int32 optimalFanOut int32
networkSize int32
requestCh chan int requestCh chan int
} }
@ -233,6 +235,11 @@ func (d *DefaultDiscovery) GetFanOut() int {
return int(atomic.LoadInt32(&d.optimalFanOut)) return int(atomic.LoadInt32(&d.optimalFanOut))
} }
// NetworkSize returns the estimated network size.
func (d *DefaultDiscovery) NetworkSize() int {
return int(atomic.LoadInt32(&d.networkSize))
}
// updateNetSize updates network size estimation metric. Must be called under read lock. // updateNetSize updates network size estimation metric. Must be called under read lock.
func (d *DefaultDiscovery) updateNetSize() { func (d *DefaultDiscovery) updateNetSize() {
var netsize = len(d.connectedAddrs) + len(d.unconnectedAddrs) + 1 // 1 for the node itself. var netsize = len(d.connectedAddrs) + len(d.unconnectedAddrs) + 1 // 1 for the node itself.
@ -242,6 +249,7 @@ func (d *DefaultDiscovery) updateNetSize() {
} }
atomic.StoreInt32(&d.optimalFanOut, int32(fanOut+0.5)) // Truncating conversion, hence +0.5. atomic.StoreInt32(&d.optimalFanOut, int32(fanOut+0.5)) // Truncating conversion, hence +0.5.
atomic.StoreInt32(&d.networkSize, int32(netsize))
updateNetworkSizeMetric(netsize) updateNetworkSizeMetric(netsize)
updatePoolCountMetric(d.poolCount()) updatePoolCountMetric(d.poolCount())
} }

View file

@ -40,6 +40,11 @@ func (d *testDiscovery) RegisterBadAddr(addr string) {
d.bad = append(d.bad, addr) d.bad = append(d.bad, addr)
} }
func (d *testDiscovery) GetFanOut() int { func (d *testDiscovery) GetFanOut() int {
d.Lock()
defer d.Unlock()
return (len(d.connected) + len(d.backfill)) * 2 / 3
}
func (d *testDiscovery) NetworkSize() int {
d.Lock() d.Lock()
defer d.Unlock() defer d.Unlock()
return len(d.connected) + len(d.backfill) return len(d.connected) + len(d.backfill)

View file

@ -387,10 +387,28 @@ func (s *Server) ConnectedPeers() []string {
// while itself dealing with peers management (handling connects/disconnects). // while itself dealing with peers management (handling connects/disconnects).
func (s *Server) run() { func (s *Server) run() {
go s.runProto() go s.runProto()
for { for loopCnt := 0; ; loopCnt++ {
if s.PeerCount() < s.MinPeers { var (
netSize = s.discovery.NetworkSize()
// "Optimal" number of peers.
optimalN = s.discovery.GetFanOut() * 2
// Real number of peers.
peerN = s.PeerCount()
)
if peerN < s.MinPeers {
// Starting up or going below the minimum -> quickly get many new peers.
s.discovery.RequestRemote(s.AttemptConnPeers) s.discovery.RequestRemote(s.AttemptConnPeers)
} else if s.MinPeers > 0 && loopCnt%s.MinPeers == 0 && optimalN > peerN && optimalN < s.MaxPeers && optimalN < netSize {
// Having some number of peers, but probably can get some more, the network is big.
// It also allows to start picking up new peers proactively, before we suddenly have <s.MinPeers of them.
var connN = s.AttemptConnPeers
if connN > optimalN-peerN {
connN = optimalN - peerN
} }
s.discovery.RequestRemote(connN)
}
if s.discovery.PoolCount() < minPoolCount { if s.discovery.PoolCount() < minPoolCount {
s.broadcastHPMessage(NewMessage(CMDGetAddr, payload.NewNullPayload())) s.broadcastHPMessage(NewMessage(CMDGetAddr, payload.NewNullPayload()))
} }