From 8627b5d128f79d720a0359884651e0c4ed41040b Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 28 Oct 2021 17:16:12 -0700 Subject: [PATCH 1/3] discovery: revamp premature update map Turns out we need it right now to handle some low latency race conditions in our integration tests, so we'll opt to simply cap the size of it to a low amount. We use a basic LRU caching mechainsm. Fixes https://github.com/lightningnetwork/lnd/issues/5076 --- discovery/gossiper.go | 62 +++++++++++++++++++++++++++++++++---------- 1 file changed, 48 insertions(+), 14 deletions(-) diff --git a/discovery/gossiper.go b/discovery/gossiper.go index 075982af8..4b98ec9df 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -12,6 +12,8 @@ import ( "github.com/btcsuite/btcd/wire" "github.com/btcsuite/btcutil" "github.com/davecgh/go-spew/spew" + "github.com/lightninglabs/neutrino/cache" + "github.com/lightninglabs/neutrino/cache/lru" "github.com/lightningnetwork/lnd/batch" "github.com/lightningnetwork/lnd/chainntnfs" "github.com/lightningnetwork/lnd/channeldb" @@ -38,6 +40,10 @@ const ( // determine how often we should allow a new update for a specific // channel and direction. DefaultChannelUpdateInterval = time.Minute + + // maxPrematureUpdates tracks the max amount of premature channel + // updates that we'll hold onto. + maxPrematureUpdates = 100 ) var ( @@ -268,6 +274,19 @@ type Config struct { ChannelUpdateInterval time.Duration } +// cachedNetworkMsg is a wrapper around a network message that can be used with +// *lru.Cache. +type cachedNetworkMsg struct { + msgs []*networkMsg +} + +// Size returns the "size" of an entry. We return the number of items as we +// just want to limit the total amount of entires rather than do accurate size +// accounting. +func (c *cachedNetworkMsg) Size() (uint64, error) { + return uint64(len(c.msgs)), nil +} + // AuthenticatedGossiper is a subsystem which is responsible for receiving // announcements, validating them and applying the changes to router, syncing // lightning network with newly connected nodes, broadcasting announcements @@ -302,8 +321,7 @@ type AuthenticatedGossiper struct { // that wasn't associated with any channel we know about. We store // them temporarily, such that we can reprocess them when a // ChannelAnnouncement for the channel is received. - prematureChannelUpdates map[uint64][]*networkMsg - pChanUpdMtx sync.Mutex + prematureChannelUpdates *lru.Cache // networkMsgs is a channel that carries new network broadcasted // message from outside the gossiper service to be processed by the @@ -368,7 +386,7 @@ func New(cfg Config, selfKeyDesc *keychain.KeyDescriptor) *AuthenticatedGossiper networkMsgs: make(chan *networkMsg), quit: make(chan struct{}), chanPolicyUpdates: make(chan *chanPolicyUpdateRequest), - prematureChannelUpdates: make(map[uint64][]*networkMsg), + prematureChannelUpdates: lru.NewCache(maxPrematureUpdates), channelMtx: multimutex.NewMutex(), recentRejects: make(map[uint64]struct{}), chanUpdateRateLimiter: make(map[uint64][2]*rate.Limiter), @@ -1774,13 +1792,14 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement( shortChanID := msg.ShortChannelID.ToUint64() var channelUpdates []*networkMsg - d.pChanUpdMtx.Lock() - channelUpdates = append(channelUpdates, d.prematureChannelUpdates[shortChanID]...) - - // Now delete the premature ChannelUpdates, since we added them - // all to the queue of network messages. - delete(d.prematureChannelUpdates, shortChanID) - d.pChanUpdMtx.Unlock() + earlyChanUpdates, err := d.prematureChannelUpdates.Get(shortChanID) + if err == nil { + // There was actually an entry in the map, so we'll + // accumulate it. We don't worry about deletion, since + // it'll eventually fall out anyway. + chanMsgs := earlyChanUpdates.(*cachedNetworkMsg) + channelUpdates = append(channelUpdates, chanMsgs.msgs...) + } // Launch a new goroutine to handle each ChannelUpdate, this to // ensure we don't block here, as we can handle only one @@ -1929,11 +1948,26 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement( // of this, we temporarily add it to a map, and // reprocess it after our own ChannelAnnouncement has // been processed. - d.pChanUpdMtx.Lock() - d.prematureChannelUpdates[shortChanID] = append( - d.prematureChannelUpdates[shortChanID], nMsg, + earlyMsgs, err := d.prematureChannelUpdates.Get( + shortChanID, ) - d.pChanUpdMtx.Unlock() + switch { + // Nothing in the cache yeyt, we can just directly + // insert this element. + case err == cache.ErrElementNotFound: + d.prematureChannelUpdates.Put(shortChanID, &cachedNetworkMsg{ + msgs: []*networkMsg{nMsg}, + }) + + // There's already something in the cache, so we'll + // combine the set of messagesa into a single value. + default: + msgs := earlyMsgs.(*cachedNetworkMsg).msgs + msgs = append(msgs, nMsg) + d.prematureChannelUpdates.Put(shortChanID, &cachedNetworkMsg{ + msgs: msgs, + }) + } log.Debugf("Got ChannelUpdate for edge not found in "+ "graph(shortChanID=%v), saving for "+ From 5a8255550bd28f97a6a28df78204f46581b2f00c Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 3 Nov 2021 18:58:03 -0700 Subject: [PATCH 2/3] discovery: revamp recent rejects map Similar to the prior commit, in this commit, we move to using a basic LRU cache to store the set of prior rejected messages. --- discovery/gossiper.go | 136 +++++++++++++++++++++++++++++------------- 1 file changed, 96 insertions(+), 40 deletions(-) diff --git a/discovery/gossiper.go b/discovery/gossiper.go index 4b98ec9df..89b72ca32 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -44,6 +44,11 @@ const ( // maxPrematureUpdates tracks the max amount of premature channel // updates that we'll hold onto. maxPrematureUpdates = 100 + + // maxRejectedUpdates tracks the max amount of rejected channel updates + // we'll maintain. This is the global size across all peers. We'll + // allocate ~3 MB max to the cache. + maxRejectedUpdates = 10_000 ) var ( @@ -287,6 +292,33 @@ func (c *cachedNetworkMsg) Size() (uint64, error) { return uint64(len(c.msgs)), nil } +// rejectCacheKey is the cache key that we'll use to track announcements we've +// recently rejected. +type rejectCacheKey struct { + pubkey [33]byte + chanID uint64 +} + +// newRejectCacheKey returns a new cache key for the reject cache. +func newRejectCacheKey(cid uint64, pub [33]byte) rejectCacheKey { + k := rejectCacheKey{ + chanID: cid, + pubkey: pub, + } + + return k +} + +// cachedReject is the empty value used to track the value for rejects. +type cachedReject struct { +} + +// Size returns the "size" of an entry. We return 1 as we just want to limit +// the total size. +func (c *cachedReject) Size() (uint64, error) { + return 1, nil +} + // AuthenticatedGossiper is a subsystem which is responsible for receiving // announcements, validating them and applying the changes to router, syncing // lightning network with newly connected nodes, broadcasting announcements @@ -345,8 +377,7 @@ type AuthenticatedGossiper struct { // consistent between when the DB is first read until it's written. channelMtx *multimutex.Mutex - rejectMtx sync.RWMutex - recentRejects map[uint64]struct{} + recentRejects *lru.Cache // syncMgr is a subsystem responsible for managing the gossip syncers // for peers currently connected. When a new peer is connected, the @@ -388,7 +419,7 @@ func New(cfg Config, selfKeyDesc *keychain.KeyDescriptor) *AuthenticatedGossiper chanPolicyUpdates: make(chan *chanPolicyUpdateRequest), prematureChannelUpdates: lru.NewCache(maxPrematureUpdates), channelMtx: multimutex.NewMutex(), - recentRejects: make(map[uint64]struct{}), + recentRejects: lru.NewCache(maxRejectedUpdates), chanUpdateRateLimiter: make(map[uint64][2]*rate.Limiter), } @@ -1036,7 +1067,9 @@ func (d *AuthenticatedGossiper) networkHandler() { // If this message was recently rejected, then we won't // attempt to re-process it. - if d.isRecentlyRejectedMsg(announcement.msg) { + if announcement.isRemote && d.isRecentlyRejectedMsg( + announcement.msg, announcement.peer.PubKey(), + ) { announcement.err <- fmt.Errorf("recently " + "rejected") continue @@ -1202,22 +1235,23 @@ func (d *AuthenticatedGossiper) PruneSyncState(peer route.Vertex) { // isRecentlyRejectedMsg returns true if we recently rejected a message, and // false otherwise, This avoids expensive reprocessing of the message. -func (d *AuthenticatedGossiper) isRecentlyRejectedMsg(msg lnwire.Message) bool { - d.rejectMtx.RLock() - defer d.rejectMtx.RUnlock() +func (d *AuthenticatedGossiper) isRecentlyRejectedMsg(msg lnwire.Message, + peerPub [33]byte) bool { + var scid uint64 switch m := msg.(type) { case *lnwire.ChannelUpdate: - _, ok := d.recentRejects[m.ShortChannelID.ToUint64()] - return ok + scid = m.ShortChannelID.ToUint64() case *lnwire.ChannelAnnouncement: - _, ok := d.recentRejects[m.ShortChannelID.ToUint64()] - return ok + scid = m.ShortChannelID.ToUint64() default: return false } + + _, err := d.recentRejects.Get(newRejectCacheKey(scid, peerPub)) + return err != cache.ErrElementNotFound } // retransmitStaleAnns examines all outgoing channels that the source node is @@ -1640,9 +1674,11 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement( d.cfg.ChainHash) log.Errorf(err.Error()) - d.rejectMtx.Lock() - d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{} - d.rejectMtx.Unlock() + key := newRejectCacheKey( + msg.ShortChannelID.ToUint64(), + nMsg.peer.PubKey(), + ) + _, _ = d.recentRejects.Put(key, &cachedReject{}) nMsg.err <- err return nil, false @@ -1680,9 +1716,12 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement( if err := routing.ValidateChannelAnn(msg); err != nil { err := fmt.Errorf("unable to validate "+ "announcement: %v", err) - d.rejectMtx.Lock() - d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{} - d.rejectMtx.Unlock() + + key := newRejectCacheKey( + msg.ShortChannelID.ToUint64(), + nMsg.peer.PubKey(), + ) + _, _ = d.recentRejects.Put(key, &cachedReject{}) log.Error(err) nMsg.err <- err @@ -1753,9 +1792,13 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement( // see if we get any new announcements. anns, rErr := d.processRejectedEdge(msg, proof) if rErr != nil { - d.rejectMtx.Lock() - d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{} - d.rejectMtx.Unlock() + + key := newRejectCacheKey( + msg.ShortChannelID.ToUint64(), + nMsg.peer.PubKey(), + ) + _, _ = d.recentRejects.Put(key, &cachedReject{}) + nMsg.err <- rErr return nil, false } @@ -1777,9 +1820,11 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement( log.Tracef("Router rejected channel "+ "edge: %v", err) - d.rejectMtx.Lock() - d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{} - d.rejectMtx.Unlock() + key := newRejectCacheKey( + msg.ShortChannelID.ToUint64(), + nMsg.peer.PubKey(), + ) + _, _ = d.recentRejects.Put(key, &cachedReject{}) } nMsg.err <- err @@ -1862,9 +1907,11 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement( d.cfg.ChainHash) log.Errorf(err.Error()) - d.rejectMtx.Lock() - d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{} - d.rejectMtx.Unlock() + key := newRejectCacheKey( + msg.ShortChannelID.ToUint64(), + nMsg.peer.PubKey(), + ) + _, _ = d.recentRejects.Put(key, &cachedReject{}) nMsg.err <- err return nil, false @@ -1952,21 +1999,23 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement( shortChanID, ) switch { - // Nothing in the cache yeyt, we can just directly + // Nothing in the cache yet, we can just directly // insert this element. case err == cache.ErrElementNotFound: - d.prematureChannelUpdates.Put(shortChanID, &cachedNetworkMsg{ - msgs: []*networkMsg{nMsg}, - }) + _, _ = d.prematureChannelUpdates.Put( + shortChanID, &cachedNetworkMsg{ + msgs: []*networkMsg{nMsg}, + }) // There's already something in the cache, so we'll - // combine the set of messagesa into a single value. + // combine the set of messages into a single value. default: msgs := earlyMsgs.(*cachedNetworkMsg).msgs msgs = append(msgs, nMsg) - d.prematureChannelUpdates.Put(shortChanID, &cachedNetworkMsg{ - msgs: msgs, - }) + _, _ = d.prematureChannelUpdates.Put( + shortChanID, &cachedNetworkMsg{ + msgs: msgs, + }) } log.Debugf("Got ChannelUpdate for edge not found in "+ @@ -1984,9 +2033,12 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement( log.Error(err) nMsg.err <- err - d.rejectMtx.Lock() - d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{} - d.rejectMtx.Unlock() + key := newRejectCacheKey( + msg.ShortChannelID.ToUint64(), + nMsg.peer.PubKey(), + ) + _, _ = d.recentRejects.Put(key, &cachedReject{}) + return nil, false } @@ -2089,9 +2141,13 @@ func (d *AuthenticatedGossiper) processNetworkAnnouncement( routing.ErrIgnored) { log.Debug(err) } else if err != routing.ErrVBarrierShuttingDown { - d.rejectMtx.Lock() - d.recentRejects[msg.ShortChannelID.ToUint64()] = struct{}{} - d.rejectMtx.Unlock() + + key := newRejectCacheKey( + msg.ShortChannelID.ToUint64(), + nMsg.peer.PubKey(), + ) + _, _ = d.recentRejects.Put(key, &cachedReject{}) + log.Error(err) } From 75ae7e68b0be34633f43866e8bdcfd2182bd3b60 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 3 Nov 2021 19:01:35 -0700 Subject: [PATCH 3/3] docs/release-notes: add new entry for gossip cache revamp --- docs/release-notes/release-notes-0.14.0.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/release-notes/release-notes-0.14.0.md b/docs/release-notes/release-notes-0.14.0.md index 757090b42..58b4df151 100644 --- a/docs/release-notes/release-notes-0.14.0.md +++ b/docs/release-notes/release-notes-0.14.0.md @@ -367,6 +367,8 @@ messages directly. There is no routing/path finding involved. * Using `go get` to install go executables is now deprecated. Migrate to `go install` our lnrpc proto dockerfile [Migrate `go get` to `go install`](https://github.com/lightningnetwork/lnd/pull/5879) +* [The premature update map has been revamped using an LRU cache](https://github.com/lightningnetwork/lnd/pull/5902) + ## Code Health ### Code cleanup, refactor, typo fixes