status-im/go-libp2p-pubsub
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

more efficient promise processing on throttling

Browse Source
This commit is contained in:
vyzo 2020-09-02 21:51:56 +03:00
parent 93f4f91dcf
commit f7f6f59758
1 changed files with 40 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
gossip_tracer.go
View File

@ -14,14 +14,16 @@ import (
// The tracking of promises is probabilistic to avoid using too much memory. // The tracking of promises is probabilistic to avoid using too much memory.
type gossipTracer struct { type gossipTracer struct {
sync.Mutex sync.Mutex
msgID MsgIdFunction msgID MsgIdFunction
promises map[string]map[peer.ID]time.Time promises map[string]map[peer.ID]time.Time
peerPromises map[peer.ID]map[string]struct{}
} }
func newGossipTracer() *gossipTracer { func newGossipTracer() *gossipTracer {
return &gossipTracer{ return &gossipTracer{
msgID: DefaultMsgIdFn, msgID: DefaultMsgIdFn,
promises: make(map[string]map[peer.ID]time.Time), promises: make(map[string]map[peer.ID]time.Time),
peerPromises: make(map[peer.ID]map[string]struct{}),
} }
} }
@ -45,15 +47,21 @@ func (gt *gossipTracer) AddPromise(p peer.ID, msgIDs []string) {
gt.Lock() gt.Lock()
defer gt.Unlock() defer gt.Unlock()
peers, ok := gt.promises[mid] promises, ok := gt.promises[mid]
if !ok { if !ok {
peers = make(map[peer.ID]time.Time) promises = make(map[peer.ID]time.Time)
gt.promises[mid] = peers gt.promises[mid] = promises
} }
_, ok = peers[p] _, ok = promises[p]
if !ok { if !ok {
peers[p] = time.Now().Add(GossipSubIWantFollowupTime) promises[p] = time.Now().Add(GossipSubIWantFollowupTime)
peerPromises, ok := gt.peerPromises[p]
if !ok {
peerPromises = make(map[string]struct{})
gt.peerPromises[p] = peerPromises
}
peerPromises[mid] = struct{}{}
} }
} }
@ -71,18 +79,25 @@ func (gt *gossipTracer) GetBrokenPromises() map[peer.ID]int {
now := time.Now() now := time.Now()
// find broken promises from peers // find broken promises from peers
for mid, peers := range gt.promises { for mid, promises := range gt.promises {
for p, expire := range peers { for p, expire := range promises {
if expire.Before(now) { if expire.Before(now) {
if res == nil { if res == nil {
res = make(map[peer.ID]int) res = make(map[peer.ID]int)
} }
res[p]++ res[p]++
delete(peers, p) delete(promises, p)
peerPromises := gt.peerPromises[p]
delete(peerPromises, mid)
if len(peerPromises) == 0 {
delete(gt.peerPromises, p)
}
} }
} }
if len(peers) == 0 {
if len(promises) == 0 {
delete(gt.promises, mid) delete(gt.promises, mid)
} }
} }
@ -140,14 +155,18 @@ func (gt *gossipTracer) ThrottlePeer(p peer.ID) {
gt.Lock() gt.Lock()
defer gt.Unlock() defer gt.Unlock()
// remove promises for peers that have been throttled so that they are not unfairly penalized peerPromises, ok := gt.peerPromises[p]
for mid, peers := range gt.promises { if !ok {
_, hasPromise := peers[p] return
if hasPromise { }
delete(peers, p)
if len(peers) == 0 { for mid := range peerPromises {
delete(gt.promises, mid) promises := gt.promises[mid]
} delete(promises, p)
if len(promises) == 0 {
delete(gt.promises, mid)
} }
} }
delete(gt.peerPromises, p)
} }