graph/db: let ForEachNodeCacheable take a context
This commit is contained in:
5 files changed
+11
-12
No files matched your search
+3
-3
@@ -75,7 +75,7 @@ func (c *ChannelGraph) Start() error {
|
|||||||
defer log.Debug("ChannelGraph started")
|
defer log.Debug("ChannelGraph started")
|
||||||
|
|
||||||
if c.graphCache != nil {
|
if c.graphCache != nil {
|
||||||
if err := c.populateCache(); err != nil {
|
if err := c.populateCache(context.TODO()); err != nil {
|
||||||
return fmt.Errorf("could not populate the graph "+
|
return fmt.Errorf("could not populate the graph "+
|
||||||
"cache: %w", err)
|
"cache: %w", err)
|
||||||
}
|
}
|
||||||
@@ -159,12 +159,12 @@ func (c *ChannelGraph) handleTopologySubscriptions() {
|
|||||||
// populateCache loads the entire channel graph into the in-memory graph cache.
|
// populateCache loads the entire channel graph into the in-memory graph cache.
|
||||||
//
|
//
|
||||||
// NOTE: This should only be called if the graphCache has been constructed.
|
// NOTE: This should only be called if the graphCache has been constructed.
|
||||||
func (c *ChannelGraph) populateCache() error {
|
func (c *ChannelGraph) populateCache(ctx context.Context) error {
|
||||||
startTime := time.Now()
|
startTime := time.Now()
|
||||||
log.Info("Populating in-memory channel graph, this might take a " +
|
log.Info("Populating in-memory channel graph, this might take a " +
|
||||||
"while...")
|
"while...")
|
||||||
|
|
||||||
err := c.V1Store.ForEachNodeCacheable(func(node route.Vertex,
|
err := c.V1Store.ForEachNodeCacheable(ctx, func(node route.Vertex,
|
||||||
features *lnwire.FeatureVector) error {
|
features *lnwire.FeatureVector) error {
|
||||||
|
|
||||||
c.graphCache.AddNodeFeatures(node, features)
|
c.graphCache.AddNodeFeatures(node, features)
|
||||||
|
|||||||
@@ -1433,6 +1433,7 @@ func TestGraphTraversal(t *testing.T) {
|
|||||||
// working correctly.
|
// working correctly.
|
||||||
func TestGraphTraversalCacheable(t *testing.T) {
|
func TestGraphTraversalCacheable(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
graph := MakeTestGraph(t)
|
graph := MakeTestGraph(t)
|
||||||
|
|
||||||
@@ -1458,7 +1459,7 @@ func TestGraphTraversalCacheable(t *testing.T) {
|
|||||||
// iterating over each node, once again if the map is empty that
|
// iterating over each node, once again if the map is empty that
|
||||||
// indicates that all edges have properly been reached.
|
// indicates that all edges have properly been reached.
|
||||||
var nodes []route.Vertex
|
var nodes []route.Vertex
|
||||||
err = graph.ForEachNodeCacheable(func(node route.Vertex,
|
err = graph.ForEachNodeCacheable(ctx, func(node route.Vertex,
|
||||||
features *lnwire.FeatureVector) error {
|
features *lnwire.FeatureVector) error {
|
||||||
|
|
||||||
delete(nodeMap, node)
|
delete(nodeMap, node)
|
||||||
@@ -4226,7 +4227,7 @@ func BenchmarkForEachChannel(b *testing.B) {
|
|||||||
)
|
)
|
||||||
|
|
||||||
var nodes []route.Vertex
|
var nodes []route.Vertex
|
||||||
err := graph.ForEachNodeCacheable(func(node route.Vertex,
|
err := graph.ForEachNodeCacheable(ctx, func(node route.Vertex,
|
||||||
vector *lnwire.FeatureVector) error {
|
vector *lnwire.FeatureVector) error {
|
||||||
|
|
||||||
nodes = append(nodes, node)
|
nodes = append(nodes, node)
|
||||||
|
|||||||
@@ -107,7 +107,7 @@ type V1Store interface { //nolint:interfacebloat
|
|||||||
// in the graph, executing the passed callback with each node
|
// in the graph, executing the passed callback with each node
|
||||||
// encountered. If the callback returns an error, then the transaction
|
// encountered. If the callback returns an error, then the transaction
|
||||||
// is aborted and the iteration stops early.
|
// is aborted and the iteration stops early.
|
||||||
ForEachNodeCacheable(cb func(route.Vertex,
|
ForEachNodeCacheable(ctx context.Context, cb func(route.Vertex,
|
||||||
*lnwire.FeatureVector) error) error
|
*lnwire.FeatureVector) error) error
|
||||||
|
|
||||||
// LookupAlias attempts to return the alias as advertised by the target
|
// LookupAlias attempts to return the alias as advertised by the target
|
||||||
|
|||||||
@@ -842,8 +842,8 @@ func forEachNode(db kvdb.Backend,
|
|||||||
// graph, executing the passed callback with each node encountered. If the
|
// graph, executing the passed callback with each node encountered. If the
|
||||||
// callback returns an error, then the transaction is aborted and the iteration
|
// callback returns an error, then the transaction is aborted and the iteration
|
||||||
// stops early.
|
// stops early.
|
||||||
func (c *KVStore) ForEachNodeCacheable(cb func(route.Vertex,
|
func (c *KVStore) ForEachNodeCacheable(_ context.Context,
|
||||||
*lnwire.FeatureVector) error) error {
|
cb func(route.Vertex, *lnwire.FeatureVector) error) error {
|
||||||
|
|
||||||
traversal := func(tx kvdb.RTx) error {
|
traversal := func(tx kvdb.RTx) error {
|
||||||
// First grab the nodes bucket which stores the mapping from
|
// First grab the nodes bucket which stores the mapping from
|
||||||
|
|||||||
@@ -923,10 +923,8 @@ func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex,
|
|||||||
// stops early.
|
// stops early.
|
||||||
//
|
//
|
||||||
// NOTE: This is a part of the V1Store interface.
|
// NOTE: This is a part of the V1Store interface.
|
||||||
func (s *SQLStore) ForEachNodeCacheable(cb func(route.Vertex,
|
func (s *SQLStore) ForEachNodeCacheable(ctx context.Context,
|
||||||
*lnwire.FeatureVector) error) error {
|
cb func(route.Vertex, *lnwire.FeatureVector) error) error {
|
||||||
|
|
||||||
ctx := context.TODO()
|
|
||||||
|
|
||||||
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
||||||
return forEachNodeCacheable(ctx, db, func(nodeID int64,
|
return forEachNodeCacheable(ctx, db, func(nodeID int64,
|
||||||
|
|||||||
Reference in new issue
Block a user