From fae470293fd8d81be5d07e7dd7d49cc65ad86842 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Tue, 19 Oct 2021 16:04:23 -0700 Subject: [PATCH 1/4] channeldb: add new ForEachNode method using the channel graph cache This commit, adds a new ForEachNode method to the channel graph cache that assumes the contents won't be modified. This is generally useful, and will be used in a later commit to optimize some heavy RPC calls. --- channeldb/graph.go | 66 +++++++++++++++++++++++++++++++++++ channeldb/graph_cache.go | 24 +++++++++++++ channeldb/graph_cache_test.go | 18 ++++++++++ channeldb/graph_test.go | 28 +++++++++++++++ 4 files changed, 136 insertions(+) diff --git a/channeldb/graph.go b/channeldb/graph.go index cd746adc7..118f1882d 100644 --- a/channeldb/graph.go +++ b/channeldb/graph.go @@ -458,6 +458,72 @@ func (c *ChannelGraph) FetchNodeFeatures( } } +// ForEachNodeCached is similar to ForEachNode, but it utilizes the channel +// graph cache instead. Note that this doesn't return all the information the +// regular ForEachNode method does. +// +// NOTE: The callback contents MUST not be modified. +func (c *ChannelGraph) ForEachNodeCached(cb func(node route.Vertex, + chans map[uint64]*DirectedChannel) error) error { + + if c.graphCache != nil { + return c.graphCache.ForEachNode(cb) + } + + // Otherwise call back to a version that uses the database directly. + // We'll iterate over each node, then the set of channels for each + // node, and construct a similar callback functiopn signature as the + // main funcotin expects. + return c.ForEachNode(func(tx kvdb.RTx, node *LightningNode) error { + channels := make(map[uint64]*DirectedChannel) + + err := node.ForEachChannel(tx, func(tx kvdb.RTx, + e *ChannelEdgeInfo, p1 *ChannelEdgePolicy, + p2 *ChannelEdgePolicy) error { + + toNodeCallback := func() route.Vertex { + return node.PubKeyBytes + } + toNodeFeatures, err := c.FetchNodeFeatures( + node.PubKeyBytes, + ) + if err != nil { + return err + } + + var cachedInPolicy *CachedEdgePolicy + if p2 != nil { + cachedInPolicy := NewCachedPolicy(p2) + cachedInPolicy.ToNodePubKey = toNodeCallback + cachedInPolicy.ToNodeFeatures = toNodeFeatures + } + + directedChannel := &DirectedChannel{ + ChannelID: e.ChannelID, + IsNode1: node.PubKeyBytes == e.NodeKey1Bytes, + OtherNode: e.NodeKey2Bytes, + Capacity: e.Capacity, + OutPolicySet: p1 != nil, + InPolicy: cachedInPolicy, + } + + if node.PubKeyBytes == e.NodeKey2Bytes { + directedChannel.OtherNode = e.NodeKey1Bytes + } + + channels[e.ChannelID] = directedChannel + + return nil + + }) + if err != nil { + return err + } + + return cb(node.PubKeyBytes, channels) + }) +} + // DisabledChannelIDs returns the channel ids of disabled channels. // A channel is disabled when two of the associated ChanelEdgePolicies // have their disabled bit on. diff --git a/channeldb/graph_cache.go b/channeldb/graph_cache.go index 891004bfc..3618b6d43 100644 --- a/channeldb/graph_cache.go +++ b/channeldb/graph_cache.go @@ -447,6 +447,30 @@ func (c *GraphCache) ForEachChannel(node route.Vertex, return nil } +// ForEachNode iterates over the adjacency list of the graph, executing the +// call back for each node and the set of channels that emanate from the given +// node. +// +// NOTE: This method should be considered _read only_, the channels or nodes +// passed in MUST NOT be modified. +func (c *GraphCache) ForEachNode(cb func(node route.Vertex, + channels map[uint64]*DirectedChannel) error) error { + + c.mtx.RLock() + defer c.mtx.RUnlock() + + for node, channels := range c.nodeChannels { + // We don't make a copy here since this is a read-only RPC + // call. We also don't need the node features either for this + // call. + if err := cb(node, channels); err != nil { + return err + } + } + + return nil +} + // GetFeatures returns the features of the node with the given ID. If no // features are known for the node, an empty feature vector is returned. func (c *GraphCache) GetFeatures(node route.Vertex) *lnwire.FeatureVector { diff --git a/channeldb/graph_cache_test.go b/channeldb/graph_cache_test.go index f8d2b9a5f..b408ec36d 100644 --- a/channeldb/graph_cache_test.go +++ b/channeldb/graph_cache_test.go @@ -120,6 +120,24 @@ func TestGraphCacheAddNode(t *testing.T) { require.Equal(t, inPolicy1 != nil, toChannels[0].OutPolicySet) assertCachedPolicyEqual(t, outPolicy1, toChannels[0].InPolicy) + + // Now that we've inserted two nodes into the graph, check that + // we'll recover the same set of channels during ForEachNode. + nodes := make(map[route.Vertex]struct{}) + chans := make(map[uint64]struct{}) + _ = cache.ForEachNode(func(node route.Vertex, + edges map[uint64]*DirectedChannel) error { + + nodes[node] = struct{}{} + for chanID := range edges { + chans[chanID] = struct{}{} + } + + return nil + }) + + require.Len(t, nodes, 2) + require.Len(t, chans, 1) } runTest(pubKey1, pubKey2) diff --git a/channeldb/graph_test.go b/channeldb/graph_test.go index eb81026c1..36ee20095 100644 --- a/channeldb/graph_test.go +++ b/channeldb/graph_test.go @@ -1103,6 +1103,34 @@ func TestGraphTraversal(t *testing.T) { const numChannels = 5 chanIndex, nodeList := fillTestGraph(t, graph, numNodes, numChannels) + // Make an index of the node list for easy look up below. + nodeIndex := make(map[route.Vertex]struct{}) + for _, node := range nodeList { + nodeIndex[node.PubKeyBytes] = struct{}{} + } + + // If we turn the channel graph cache _off_, then iterate through the + // set of channels (to force the fall back), we should find all the + // channel as well as the nodes included. + graph.graphCache = nil + err = graph.ForEachNodeCached(func(node route.Vertex, + chans map[uint64]*DirectedChannel) error { + + if _, ok := nodeIndex[node]; !ok { + return fmt.Errorf("node %x not found in graph", node) + } + + for chanID := range chans { + if _, ok := chanIndex[chanID]; !ok { + return fmt.Errorf("chan %v not found in "+ + "graph", chanID) + } + } + + return nil + }) + require.NoError(t, err) + // Iterate through all the known channels within the graph DB, once // again if the map is empty that indicates that all edges have // properly been reached. From 79406696ad574858e0b108807a9728c9567fc544 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Tue, 19 Oct 2021 16:04:51 -0700 Subject: [PATCH 2/4] rpc: optimize GetNetworkInfo by using the channel graph cache --- rpcserver.go | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/rpcserver.go b/rpcserver.go index 9b15348ab..5079f86d2 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -5708,7 +5708,9 @@ func (r *rpcServer) GetNetworkInfo(ctx context.Context, // network, tallying up the total number of nodes, and also gathering // each node so we can measure the graph diameter and degree stats // below. - if err := graph.ForEachNode(func(tx kvdb.RTx, node *channeldb.LightningNode) error { + err := graph.ForEachNodeCached(func(node route.Vertex, + edges map[uint64]*channeldb.DirectedChannel) error { + // Increment the total number of nodes with each iteration. numNodes++ @@ -5718,9 +5720,7 @@ func (r *rpcServer) GetNetworkInfo(ctx context.Context, // through the db transaction from the outer view so we can // re-use it within this inner view. var outDegree uint32 - if err := node.ForEachChannel(tx, func(_ kvdb.RTx, - edge *channeldb.ChannelEdgeInfo, _, _ *channeldb.ChannelEdgePolicy) error { - + for _, edge := range edges { // Bump up the out degree for this node for each // channel encountered. outDegree++ @@ -5751,9 +5751,6 @@ func (r *rpcServer) GetNetworkInfo(ctx context.Context, seenChans[edge.ChannelID] = struct{}{} allChans = append(allChans, edge.Capacity) - return nil - }); err != nil { - return err } // Finally, if the out degree of this node is greater than what @@ -5763,7 +5760,8 @@ func (r *rpcServer) GetNetworkInfo(ctx context.Context, } return nil - }); err != nil { + }) + if err != nil { return nil, err } From 67e18e6089b787343130ca814dc51757d6b87d8c Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Tue, 19 Oct 2021 16:19:53 -0700 Subject: [PATCH 3/4] rpc: optimize DescribeGraph by caching the response Unfortunately, we can't use the graph cache directly here as it doesn't include all the information we need, since it only includes the minimal amount of information needed for path finding. Instead, we use a simple mutex guarded cache that evicts itself after a certain interval. The default is set small enough that most users shouldn't really notice. We also provide a way to disable the cache, and also disable the cache in the itests. Fixes https://github.com/lightningnetwork/lnd/issues/1232 --- lncfg/caches.go | 13 ++++++++++++- lntest/node.go | 1 + rpcserver.go | 42 ++++++++++++++++++++++++++++++++++++++++++ sample-lnd.conf | 4 ++++ 4 files changed, 59 insertions(+), 1 deletion(-) diff --git a/lncfg/caches.go b/lncfg/caches.go index 47b9a97e8..abb1cf24d 100644 --- a/lncfg/caches.go +++ b/lncfg/caches.go @@ -1,6 +1,9 @@ package lncfg -import "fmt" +import ( + "fmt" + "time" +) const ( // MinRejectCacheSize is a floor on the maximum capacity allowed for @@ -10,6 +13,10 @@ const ( // MinChannelCacheSize is a floor on the maximum capacity allowed for // channeldb's channel cache. This amounts to roughly 2 MB when full. MinChannelCacheSize = 1000 + + // DefaultRPCGraphCacheDuration is the default interval that the RPC + // response to DescribeGraph should be cached for. + DefaultRPCGraphCacheDuration = time.Minute ) // Caches holds the configuration for various caches within lnd. @@ -24,6 +31,10 @@ type Caches struct { // peers querying for gossip traffic. Memory usage is roughly 2Kb per // entry. ChannelCacheSize int `long:"channel-cache-size" description:"Maximum number of entries contained in the channel cache, which is used to reduce memory allocations from gossip queries from peers. Each entry requires roughly 2Kb."` + + // RPCGraphCacheDuration is used to control the flush interval of the + // channel graph cache. + RPCGraphCacheDuration time.Duration `long:"rpc-graph-cache-duration" description:"The period of time expressed as a duration (1s, 1m, 1h, etc) that the RPC response to DescribeGraph should be cached for."` } // Validate checks the Caches configuration for values that are too small to be diff --git a/lntest/node.go b/lntest/node.go index f164ad070..f56fe1c91 100644 --- a/lntest/node.go +++ b/lntest/node.go @@ -303,6 +303,7 @@ func (cfg NodeConfig) genArgs() []string { args = append(args, fmt.Sprintf("--invoicemacaroonpath=%v", cfg.InvoiceMacPath)) args = append(args, fmt.Sprintf("--trickledelay=%v", trickleDelay)) args = append(args, fmt.Sprintf("--profile=%d", cfg.ProfilePort)) + args = append(args, fmt.Sprintf("--caches.rpc-graph-cache-duration=0")) if !cfg.HasSeed { args = append(args, "--noseedbackup") diff --git a/rpcserver.go b/rpcserver.go index 5079f86d2..674e64d51 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -628,6 +628,10 @@ type rpcServer struct { // interceptor is used to be able to request a shutdown interceptor signal.Interceptor + + graphCache sync.RWMutex + describeGraphResp *lnrpc.ChannelGraph + graphCacheEvictor *time.Timer } // A compile time check to ensure that rpcServer fully implements the @@ -813,6 +817,23 @@ func (r *rpcServer) addDeps(s *server, macService *macaroons.Service, r.chanPredicate = chanPredicate r.macService = macService r.selfNode = selfNode.PubKeyBytes + + graphCacheDuration := r.cfg.Caches.RPCGraphCacheDuration + if graphCacheDuration != 0 { + r.graphCacheEvictor = time.AfterFunc(graphCacheDuration, func() { + // Grab the mutex and purge the current populated + // describe graph response. + r.graphCache.Lock() + defer r.graphCache.Unlock() + + r.describeGraphResp = nil + + // Reset ourselves as well at the end so we run again + // after the duration. + r.graphCacheEvictor.Reset(graphCacheDuration) + }) + } + return nil } @@ -5381,6 +5402,20 @@ func (r *rpcServer) DescribeGraph(ctx context.Context, resp := &lnrpc.ChannelGraph{} includeUnannounced := req.IncludeUnannounced + // Check to see if the cache is already populated, if so then we can + // just return it directly. + // + // TODO(roasbeef): move this to an interceptor level feature? + graphCacheActive := r.cfg.Caches.RPCGraphCacheDuration != 0 + if graphCacheActive { + r.graphCache.Lock() + defer r.graphCache.Unlock() + + if r.describeGraphResp != nil { + return r.describeGraphResp, nil + } + } + // Obtain the pointer to the global singleton channel graph, this will // provide a consistent view of the graph due to bolt db's // transactional model. @@ -5439,6 +5474,13 @@ func (r *rpcServer) DescribeGraph(ctx context.Context, return nil, err } + // We still have the mutex held, so we can safely populate the cache + // now to save on GC churn for this query, but only if the cache isn't + // disabled. + if graphCacheActive { + r.describeGraphResp = resp + } + return resp, nil } diff --git a/sample-lnd.conf b/sample-lnd.conf index 69533d338..7601873b4 100644 --- a/sample-lnd.conf +++ b/sample-lnd.conf @@ -1088,6 +1088,10 @@ litecoin.node=ltcd ; roughly 2Kb. (default: 20000) ; caches.channel-cache-size=9000000 +; The duration that the response to DescribeGraph should be cached for. Setting +; the value to zero disables the cache. (default: 1m) +; caches.rpc-graph-cache-duration=10m + [protocol] From 090b917ee24a63749eaa96c5c228c47cf08b03b5 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Tue, 19 Oct 2021 16:25:21 -0700 Subject: [PATCH 4/4] docs/release-notes: add entry for rpc graph cache --- docs/release-notes/release-notes-0.14.0.md | 5 +++++ 1 file changed, 5 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..867e52ac5 100644 --- a/docs/release-notes/release-notes-0.14.0.md +++ b/docs/release-notes/release-notes-0.14.0.md @@ -515,6 +515,11 @@ messages directly. There is no routing/path finding involved. buffer each time we decrypt an incoming message, as we recycle these buffers in the peer. +* [The `DescribeGraph` and `GetNetworkInfo` calls have been + optimized](https://github.com/lightningnetwork/lnd/pull/5873) by caching the + response periodically, or using the new channel graph cache directly. This + should significantly cut down on the garbage these two calls generate. + ## Log system * [Save compressed log files from logrorate during