13543202610

Committed 26 Feb 2025 11:50AM UTC coverage: 58.834% (+0.009%) from 58.825%

Build # 13543202610

Build Type

Pull #9544

github

Committed by

web-flow

Commit Message

Merge pull request #9550 from ellemouton/graph14

graph: extract cache from CRUD [3]

Pull Request Pull Request #9544: graph: move graph cache out of CRUD layer

Run Details

2548 of 3279 new or added lines in 5 files covered. (77.71%)

301 existing lines in 21 files now uncovered.

136426 of 231881 relevant lines covered (58.83%)

19328.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.57

/graph/db/graph.go

package graphdb

import (
        "sync"
        "time"

        "github.com/lightningnetwork/lnd/batch"
        "github.com/lightningnetwork/lnd/graph/db/models"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/routing/route"
)

// Config is a struct that holds all the necessary dependencies for a
// ChannelGraph.
type Config struct {
        // KVDB is the kvdb.Backend that will be used for initializing the
        // KVStore CRUD layer.
        KVDB kvdb.Backend

        // KVStoreOpts is a list of functional options that will be used when
        // initializing the KVStore.
        KVStoreOpts []KVStoreOptionModifier
}

// ChannelGraph is a layer above the graph's CRUD layer.
//
// NOTE: currently, this is purely a pass-through layer directly to the backing
// KVStore. Upcoming commits will move the graph cache out of the KVStore and
// into this layer so that the KVStore is only responsible for CRUD operations.
type ChannelGraph struct {
        // cacheMu guards any writes to the graphCache. It should be held
        // across the DB write call and the graphCache update to make the
        // two updates as atomic as possible.
        cacheMu    sync.Mutex
        graphCache *GraphCache

        *KVStore
}

// NewChannelGraph creates a new ChannelGraph instance with the given backend.
func NewChannelGraph(cfg *Config, options ...ChanGraphOption) (*ChannelGraph,
        error) {

        opts := defaultChanGraphOptions()
        for _, o := range options {
                o(opts)
        }

        store, err := NewKVStore(cfg.KVDB, cfg.KVStoreOpts...)
        if err != nil {
                return nil, err
        }

        if !opts.useGraphCache {
                return &ChannelGraph{
                        KVStore: store,
                }, nil
        }

        // The graph cache can be turned off (e.g. for mobile users) for a
        // speed/memory usage tradeoff.
        graphCache := NewGraphCache(opts.preAllocCacheNumNodes)
        startTime := time.Now()
        log.Debugf("Populating in-memory channel graph, this might take a " +
                "while...")

        err = store.ForEachNodeCacheable(func(node route.Vertex,
                features *lnwire.FeatureVector) error {

                graphCache.AddNodeFeatures(node, features)

                return nil
        })
        if err != nil {
                return nil, err
        }

        err = store.ForEachChannel(func(info *models.ChannelEdgeInfo,
                policy1, policy2 *models.ChannelEdgePolicy) error {

                graphCache.AddChannel(info, policy1, policy2)

                return nil
        })
        if err != nil {
                return nil, err
        }

        log.Debugf("Finished populating in-memory channel graph (took %v, %s)",
                time.Since(startTime), graphCache.Stats())

        store.setGraphCache(graphCache)

        return &ChannelGraph{
                KVStore:    store,
                graphCache: graphCache,
        }, nil
}

// ForEachNodeDirectedChannel iterates through all channels of a given node,
// executing the passed callback on the directed edge representing the channel
// and its incoming policy. If the callback returns an error, then the iteration
// is halted with the error propagated back up to the caller. If the graphCache
// is available, then it will be used to retrieve the node's channels instead
// of the database.
//
// Unknown policies are passed into the callback as nil values.
//
// NOTE: this is part of the graphdb.NodeTraverser interface.
func (c *ChannelGraph) ForEachNodeDirectedChannel(node route.Vertex,
        cb func(channel *DirectedChannel) error) error {

        if c.graphCache != nil {
                return c.graphCache.ForEachChannel(node, cb)
        }

        return c.KVStore.ForEachNodeDirectedChannel(node, cb)
}

// FetchNodeFeatures returns the features of the given node. If no features are
// known for the node, an empty feature vector is returned.
// If the graphCache is available, then it will be used to retrieve the node's
// features instead of the database.
//
// NOTE: this is part of the graphdb.NodeTraverser interface.
func (c *ChannelGraph) FetchNodeFeatures(node route.Vertex) (
        *lnwire.FeatureVector, error) {

        if c.graphCache != nil {
                return c.graphCache.GetFeatures(node), nil
        }

        return c.KVStore.FetchNodeFeatures(node)
}

// GraphSession will provide the call-back with access to a NodeTraverser
// instance which can be used to perform queries against the channel graph. If
// the graph cache is not enabled, then the call-back will be provided with
// access to the graph via a consistent read-only transaction.
func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error) error {
        if c.graphCache != nil {
                return cb(c)
        }

        return c.KVStore.GraphSession(cb)
}

// ForEachNodeCached iterates through all the stored vertices/nodes in the
// graph, executing the passed callback with each node encountered.
//
// 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)
        }

        return c.KVStore.ForEachNodeCached(cb)
}

// AddLightningNode adds a vertex/node to the graph database. If the node is not
// in the database from before, this will add a new, unconnected one to the
// graph. If it is present from before, this will update that node's
// information. Note that this method is expected to only be called to update an
// already present node from a node announcement, or to insert a node found in a
// channel update.
func (c *ChannelGraph) AddLightningNode(node *models.LightningNode,
        op ...batch.SchedulerOption) error {

        c.cacheMu.Lock()
        defer c.cacheMu.Unlock()

        err := c.KVStore.AddLightningNode(node, op...)
        if err != nil {
                return err
        }

        if c.graphCache != nil {
                c.graphCache.AddNodeFeatures(
                        node.PubKeyBytes, node.Features,
                )
        }

        return nil
}

// DeleteLightningNode starts a new database transaction to remove a vertex/node
// from the database according to the node's public key.
func (c *ChannelGraph) DeleteLightningNode(nodePub route.Vertex) error {
        c.cacheMu.Lock()
        defer c.cacheMu.Unlock()

        err := c.KVStore.DeleteLightningNode(nodePub)
        if err != nil {
                return err
        }

        if c.graphCache != nil {
                c.graphCache.RemoveNode(nodePub)
        }

        return nil
}

// AddChannelEdge adds a new (undirected, blank) edge to the graph database. An
// undirected edge from the two target nodes are created. The information stored
// denotes the static attributes of the channel, such as the channelID, the keys
// involved in creation of the channel, and the set of features that the channel
// supports. The chanPoint and chanID are used to uniquely identify the edge
// globally within the database.
func (c *ChannelGraph) AddChannelEdge(edge *models.ChannelEdgeInfo,
        op ...batch.SchedulerOption) error {

        c.cacheMu.Lock()
        defer c.cacheMu.Unlock()

        err := c.KVStore.AddChannelEdge(edge, op...)
        if err != nil {
                return err
        }

        if c.graphCache != nil {
                c.graphCache.AddChannel(edge, nil, nil)
        }

        return nil
}

// MarkEdgeLive clears an edge from our zombie index, deeming it as live.
// If the cache is enabled, the edge will be added back to the graph cache if
// we still have a record of this channel in the DB.
func (c *ChannelGraph) MarkEdgeLive(chanID uint64) error {
        c.cacheMu.Lock()
        defer c.cacheMu.Unlock()

        err := c.KVStore.MarkEdgeLive(chanID)
        if err != nil {
                return err
        }

        if c.graphCache != nil {
                // We need to add the channel back into our graph cache,
                // otherwise we won't use it for path finding.
                infos, err := c.KVStore.FetchChanInfos([]uint64{chanID})
                if err != nil {
                        return err
                }

                if len(infos) == 0 {
                        return nil
                }

                info := infos[0]

                c.graphCache.AddChannel(info.Info, info.Policy1, info.Policy2)
        }

        return nil
}

1	package graphdb
2
3	import (
4	"sync"
5	"time"
6
7	"github.com/lightningnetwork/lnd/batch"
8	"github.com/lightningnetwork/lnd/graph/db/models"
9	"github.com/lightningnetwork/lnd/kvdb"
10	"github.com/lightningnetwork/lnd/lnwire"
11	"github.com/lightningnetwork/lnd/routing/route"
12	)
13
14	// Config is a struct that holds all the necessary dependencies for a
15	// ChannelGraph.
16	type Config struct {
17	// KVDB is the kvdb.Backend that will be used for initializing the
18	// KVStore CRUD layer.
19	KVDB kvdb.Backend
20
21	// KVStoreOpts is a list of functional options that will be used when
22	// initializing the KVStore.
23	KVStoreOpts []KVStoreOptionModifier
24	}
25
26	// ChannelGraph is a layer above the graph's CRUD layer.
27	//
28	// NOTE: currently, this is purely a pass-through layer directly to the backing
29	// KVStore. Upcoming commits will move the graph cache out of the KVStore and
30	// into this layer so that the KVStore is only responsible for CRUD operations.
31	type ChannelGraph struct {
32	// cacheMu guards any writes to the graphCache. It should be held
33	// across the DB write call and the graphCache update to make the
34	// two updates as atomic as possible.
35	cacheMu sync.Mutex
36	graphCache *GraphCache
37
38	*KVStore
39	}
40
41	// NewChannelGraph creates a new ChannelGraph instance with the given backend.
42	func NewChannelGraph(cfg Config, options ...ChanGraphOption) (ChannelGraph,
43	error) {	176✔
44		176✔
45	opts := defaultChanGraphOptions()	176✔
46	for _, o := range options {	281✔
47	o(opts)	105✔
48	}	105✔
49
50	store, err := NewKVStore(cfg.KVDB, cfg.KVStoreOpts...)	176✔
51	if err != nil {	176✔
NEW 52	return nil, err	×
UNCOV 53	}	×
54
55	if !opts.useGraphCache {	212✔
56	return &ChannelGraph{	36✔
57	KVStore: store,	36✔
58	}, nil	36✔
59	}	36✔
60
61	// The graph cache can be turned off (e.g. for mobile users) for a
62	// speed/memory usage tradeoff.
63	graphCache := NewGraphCache(opts.preAllocCacheNumNodes)	143✔
64	startTime := time.Now()	143✔
65	log.Debugf("Populating in-memory channel graph, this might take a " +	143✔
66	"while...")	143✔
67		143✔
68	err = store.ForEachNodeCacheable(func(node route.Vertex,	143✔
69	features *lnwire.FeatureVector) error {	246✔
70		103✔
71	graphCache.AddNodeFeatures(node, features)	103✔
72		103✔
73	return nil	103✔
74	})	103✔
75	if err != nil {	143✔
NEW 76	return nil, err	×
77	}	×
78
79	err = store.ForEachChannel(func(info *models.ChannelEdgeInfo,	143✔
80	policy1, policy2 *models.ChannelEdgePolicy) error {	542✔
81		399✔
82	graphCache.AddChannel(info, policy1, policy2)	399✔
83		399✔
84	return nil	399✔
85	})	399✔
86	if err != nil {	143✔
NEW 87	return nil, err	×
88	}	×
89
90	log.Debugf("Finished populating in-memory channel graph (took %v, %s)",	143✔
91	time.Since(startTime), graphCache.Stats())	143✔
92		143✔
93	store.setGraphCache(graphCache)	143✔
94		143✔
95	return &ChannelGraph{	143✔
96	KVStore: store,	143✔
97	graphCache: graphCache,	143✔
98	}, nil	143✔
99	}
100
101	// ForEachNodeDirectedChannel iterates through all channels of a given node,
102	// executing the passed callback on the directed edge representing the channel
103	// and its incoming policy. If the callback returns an error, then the iteration
104	// is halted with the error propagated back up to the caller. If the graphCache
105	// is available, then it will be used to retrieve the node's channels instead
106	// of the database.
107	//
108	// Unknown policies are passed into the callback as nil values.
109	//
110	// NOTE: this is part of the graphdb.NodeTraverser interface.
111	func (c *ChannelGraph) ForEachNodeDirectedChannel(node route.Vertex,
112	cb func(channel *DirectedChannel) error) error {	468✔
113		468✔
114	if c.graphCache != nil {	933✔
115	return c.graphCache.ForEachChannel(node, cb)	465✔
116	}	465✔
117
118	return c.KVStore.ForEachNodeDirectedChannel(node, cb)	6✔
119	}
120
121	// FetchNodeFeatures returns the features of the given node. If no features are
122	// known for the node, an empty feature vector is returned.
123	// If the graphCache is available, then it will be used to retrieve the node's
124	// features instead of the database.
125	//
126	// NOTE: this is part of the graphdb.NodeTraverser interface.
127	func (c *ChannelGraph) FetchNodeFeatures(node route.Vertex) (
128	*lnwire.FeatureVector, error) {	456✔
129		456✔
130	if c.graphCache != nil {	912✔
131	return c.graphCache.GetFeatures(node), nil	456✔
132	}	456✔
133
134	return c.KVStore.FetchNodeFeatures(node)	3✔
135	}
136
137	// GraphSession will provide the call-back with access to a NodeTraverser
138	// instance which can be used to perform queries against the channel graph. If
139	// the graph cache is not enabled, then the call-back will be provided with
140	// access to the graph via a consistent read-only transaction.
141	func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error) error {	136✔
142	if c.graphCache != nil {	218✔
143	return cb(c)	82✔
144	}	82✔
145
146	return c.KVStore.GraphSession(cb)	54✔
147	}
148
149	// ForEachNodeCached iterates through all the stored vertices/nodes in the
150	// graph, executing the passed callback with each node encountered.
151	//
152	// NOTE: The callback contents MUST not be modified.
153	func (c *ChannelGraph) ForEachNodeCached(cb func(node route.Vertex,
154	chans map[uint64]*DirectedChannel) error) error {	1✔
155		1✔
156	if c.graphCache != nil {	1✔
157	return c.graphCache.ForEachNode(cb)	×
NEW 158	}	×
159
160	return c.KVStore.ForEachNodeCached(cb)	1✔
161	}
162
163	// AddLightningNode adds a vertex/node to the graph database. If the node is not
164	// in the database from before, this will add a new, unconnected one to the
165	// graph. If it is present from before, this will update that node's
166	// information. Note that this method is expected to only be called to update an
167	// already present node from a node announcement, or to insert a node found in a
168	// channel update.
169	func (c ChannelGraph) AddLightningNode(node models.LightningNode,
170	op ...batch.SchedulerOption) error {	803✔
171		803✔
172	c.cacheMu.Lock()	803✔
173	defer c.cacheMu.Unlock()	803✔
174		803✔
175	err := c.KVStore.AddLightningNode(node, op...)	803✔
176	if err != nil {	803✔
NEW 177	return err	×
UNCOV 178	}	×
179
180	if c.graphCache != nil {	1,419✔
181	c.graphCache.AddNodeFeatures(	616✔
182	node.PubKeyBytes, node.Features,	616✔
183	)	616✔
184	}	616✔
185
186	return nil	803✔
187	}
188
189	// DeleteLightningNode starts a new database transaction to remove a vertex/node
190	// from the database according to the node's public key.
191	func (c *ChannelGraph) DeleteLightningNode(nodePub route.Vertex) error {	3✔
192	c.cacheMu.Lock()	3✔
193	defer c.cacheMu.Unlock()	3✔
194		3✔
195	err := c.KVStore.DeleteLightningNode(nodePub)	3✔
196	if err != nil {	3✔
NEW 197	return err	×
UNCOV 198	}	×
199
200	if c.graphCache != nil {	6✔
201	c.graphCache.RemoveNode(nodePub)	3✔
202	}	3✔
203
204	return nil	3✔
205	}
206
207	// AddChannelEdge adds a new (undirected, blank) edge to the graph database. An
208	// undirected edge from the two target nodes are created. The information stored
209	// denotes the static attributes of the channel, such as the channelID, the keys
210	// involved in creation of the channel, and the set of features that the channel
211	// supports. The chanPoint and chanID are used to uniquely identify the edge
212	// globally within the database.
213	func (c ChannelGraph) AddChannelEdge(edge models.ChannelEdgeInfo,
214	op ...batch.SchedulerOption) error {	1,715✔
215		1,715✔
216	c.cacheMu.Lock()	1,715✔
217	defer c.cacheMu.Unlock()	1,715✔
218		1,715✔
219	err := c.KVStore.AddChannelEdge(edge, op...)	1,715✔
220	if err != nil {	1,949✔
221	return err	234✔
222	}	234✔
223
224	if c.graphCache != nil {	2,772✔
225	c.graphCache.AddChannel(edge, nil, nil)	1,291✔
226	}	1,291✔
227
228	return nil	1,481✔
229	}
230
231	// MarkEdgeLive clears an edge from our zombie index, deeming it as live.
232	// If the cache is enabled, the edge will be added back to the graph cache if
233	// we still have a record of this channel in the DB.
234	func (c *ChannelGraph) MarkEdgeLive(chanID uint64) error {	2✔
235	c.cacheMu.Lock()	2✔
236	defer c.cacheMu.Unlock()	2✔
237		2✔
238	err := c.KVStore.MarkEdgeLive(chanID)	2✔
239	if err != nil {	3✔
240	return err	1✔
241	}	1✔
242
243	if c.graphCache != nil {	2✔
244	// We need to add the channel back into our graph cache,	1✔
245	// otherwise we won't use it for path finding.	1✔
246	infos, err := c.KVStore.FetchChanInfos([]uint64{chanID})	1✔
247	if err != nil {	1✔
NEW 248	return err	×
249	}	×
250
251	if len(infos) == 0 {	2✔
252	return nil	1✔
253	}	1✔
254
NEW 255	info := infos[0]	×
UNCOV 256		×
NEW 257	c.graphCache.AddChannel(info.Info, info.Policy1, info.Policy2)	×
258	}
259
260	return nil	×
261	}

lightningnetwork / lnd / 13543202610

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous