From b497c4694e5ddfe8908d97cb788b67f86d739054 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 19 Feb 2025 07:51:54 -0300 Subject: [PATCH] multi: add Start and Stop methods for ChannelGraph We do this in preparation for moving channel cache population logic out of the constructor and into the Start method. We also will later on (when topology subscription is moved to the ChannelGraph), have a goroutine that will need to be kicked off and stopped. --- autopilot/prefattach_test.go | 5 +++++ graph/db/graph.go | 40 +++++++++++++++++++++++++++++++++++- graph/db/graph_test.go | 8 ++++++++ graph/db/kv_store.go | 3 +++ graph/notifications_test.go | 4 ++++ peer/test_utils.go | 4 ++++ routing/pathfind_test.go | 4 ++++ server.go | 9 ++++++++ 8 files changed, 76 insertions(+), 1 deletion(-) diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index f553482bb..f20c3a480 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -49,6 +49,11 @@ func newDiskChanGraph(t *testing.T) (testGraph, error) { graphDB, err := graphdb.NewChannelGraph(&graphdb.Config{KVDB: backend}) require.NoError(t, err) + require.NoError(t, graphDB.Start()) + t.Cleanup(func() { + require.NoError(t, graphDB.Stop()) + }) + return &testDBGraph{ db: graphDB, databaseChannelGraph: databaseChannelGraph{ diff --git a/graph/db/graph.go b/graph/db/graph.go index 6e2132a5f..f926aaea3 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -3,6 +3,7 @@ package graphdb import ( "errors" "sync" + "sync/atomic" "time" "github.com/btcsuite/btcd/chaincfg/chainhash" @@ -32,13 +33,20 @@ type Config struct { // 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 { + started atomic.Bool + stopped atomic.Bool + // 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 + cacheMu sync.Mutex + graphCache *GraphCache *KVStore + + quit chan struct{} + wg sync.WaitGroup } // NewChannelGraph creates a new ChannelGraph instance with the given backend. @@ -58,6 +66,7 @@ func NewChannelGraph(cfg *Config, options ...ChanGraphOption) (*ChannelGraph, if !opts.useGraphCache { return &ChannelGraph{ KVStore: store, + quit: make(chan struct{}), }, nil } @@ -96,9 +105,38 @@ func NewChannelGraph(cfg *Config, options ...ChanGraphOption) (*ChannelGraph, return &ChannelGraph{ KVStore: store, graphCache: graphCache, + quit: make(chan struct{}), }, nil } +// Start kicks off any goroutines required for the ChannelGraph to function. +// If the graph cache is enabled, then it will be populated with the contents of +// the database. +func (c *ChannelGraph) Start() error { + if !c.started.CompareAndSwap(false, true) { + return nil + } + log.Debugf("ChannelGraph starting") + defer log.Debug("ChannelGraph started") + + return nil +} + +// Stop signals any active goroutines for a graceful closure. +func (c *ChannelGraph) Stop() error { + if !c.stopped.CompareAndSwap(false, true) { + return nil + } + + log.Debugf("ChannelGraph shutting down...") + defer log.Debug("ChannelGraph shutdown complete") + + close(c.quit) + c.wg.Wait() + + return 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 diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index bc130c599..dc41f02a2 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -4077,6 +4077,10 @@ func TestGraphLoading(t *testing.T) { graph, err := NewChannelGraph(&Config{KVDB: backend}) require.NoError(t, err) + require.NoError(t, graph.Start()) + t.Cleanup(func() { + require.NoError(t, graph.Stop()) + }) // Populate the graph with test data. const numNodes = 100 @@ -4087,6 +4091,10 @@ func TestGraphLoading(t *testing.T) { // populated. graphReloaded, err := NewChannelGraph(&Config{KVDB: backend}) require.NoError(t, err) + require.NoError(t, graphReloaded.Start()) + t.Cleanup(func() { + require.NoError(t, graphReloaded.Stop()) + }) // Assert that the cache content is identical. require.Equal( diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 0ea61a76c..aa120a39a 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -26,6 +26,7 @@ import ( "github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/routing/route" + "github.com/stretchr/testify/require" ) var ( @@ -4722,10 +4723,12 @@ func MakeTestGraph(t testing.TB, modifiers ...KVStoreOptionModifier) ( return nil, err } + require.NoError(t, graph.Start()) t.Cleanup(func() { _ = backend.Close() backendCleanup() + require.NoError(t, graph.Stop()) }) return graph, nil diff --git a/graph/notifications_test.go b/graph/notifications_test.go index aab1d4137..807b3fea7 100644 --- a/graph/notifications_test.go +++ b/graph/notifications_test.go @@ -1100,6 +1100,10 @@ func makeTestGraph(t *testing.T, useCache bool) (*graphdb.ChannelGraph, if err != nil { return nil, nil, err } + require.NoError(t, graph.Start()) + t.Cleanup(func() { + require.NoError(t, graph.Stop()) + }) return graph, backend, nil } diff --git a/peer/test_utils.go b/peer/test_utils.go index 7d3c5ca37..87a80712f 100644 --- a/peer/test_utils.go +++ b/peer/test_utils.go @@ -619,6 +619,10 @@ func createTestPeer(t *testing.T) *peerTestCtx { KVDB: graphBackend, }) require.NoError(t, err) + require.NoError(t, dbAliceGraph.Start()) + t.Cleanup(func() { + require.NoError(t, dbAliceGraph.Stop()) + }) dbAliceChannel := channeldb.OpenForTesting(t, dbPath) diff --git a/routing/pathfind_test.go b/routing/pathfind_test.go index 029383e0b..8a0280686 100644 --- a/routing/pathfind_test.go +++ b/routing/pathfind_test.go @@ -173,6 +173,10 @@ func makeTestGraph(t *testing.T, useCache bool) (*graphdb.ChannelGraph, if err != nil { return nil, nil, err } + require.NoError(t, graph.Start()) + t.Cleanup(func() { + require.NoError(t, graph.Stop()) + }) return graph, backend, nil } diff --git a/server.go b/server.go index ad521cd67..731d69797 100644 --- a/server.go +++ b/server.go @@ -2377,6 +2377,12 @@ func (s *server) Start() error { return } + cleanup = cleanup.add(s.graphDB.Stop) + if err := s.graphDB.Start(); err != nil { + startErr = err + return + } + cleanup = cleanup.add(s.graphBuilder.Stop) if err := s.graphBuilder.Start(); err != nil { startErr = err @@ -2683,6 +2689,9 @@ func (s *server) Stop() error { if err := s.graphBuilder.Stop(); err != nil { srvrLog.Warnf("failed to stop graphBuilder %v", err) } + if err := s.graphDB.Stop(); err != nil { + srvrLog.Warnf("failed to stop graphDB %v", err) + } if err := s.chainArb.Stop(); err != nil { srvrLog.Warnf("failed to stop chainArb: %v", err) }