23478841560

Committed 24 Mar 2026 07:53AM UTC coverage: 62.122% (+10.0%) from 52.142%

Build # 23478841560

Build Type

Pull #10065

github

Committed by

web-flow

Commit Message

Merge f733eed26 into db765a752

Pull Request Pull Request #10065: graph/db: async graph cache population

Coverage Stats

199 of 216 new or added lines in 6 files covered. (92.13%)

38 existing lines in 7 files now uncovered.

141310 of 227473 relevant lines covered (62.12%)

19356.05 hits per line

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

92.16

/graph/db/graph_cache_state.go

package graphdb

import (
        "sync"
        "sync/atomic"
)

// pendingUpdatesWarnThreshold is the number of buffered cache mutations at
// which a warning is logged. A large buffer indicates that cache population is
// taking a long time relative to the incoming gossip rate.
const pendingUpdatesWarnThreshold = 10_000

// graphCacheState tracks the in-memory graph cache together with its
// population state. The underlying GraphCache is independently thread-safe, so
// once reads are allowed to use it, they do not need to hold updateMtx.
type graphCacheState struct {
        graphCache *GraphCache
        loaded     atomic.Bool
        failed     atomic.Bool

        updateMtx sync.Mutex
        loading   bool

        pendingUpdates []func(*GraphCache)
}

// newGraphCacheState constructs a graph cache state with a new cache instance.
func newGraphCacheState(preAllocNumNodes int) *graphCacheState {
        return &graphCacheState{
                graphCache: NewGraphCache(preAllocNumNodes),
        }
}

// isLoaded reports whether the cache has finished its initial population and
// is safe to serve reads from.
func (s *graphCacheState) isLoaded() bool {
        return s.loaded.Load()
}

// isFailed reports whether the cache population attempt has failed.
func (s *graphCacheState) isFailed() bool {
        return s.failed.Load()
}

// stats returns the cache stats if the cache has finished its initial
// population.
func (s *graphCacheState) stats() (string, bool) {
        if !s.isLoaded() {
                return "", false
        }

        return s.graphCache.Stats(), true
}

// beginPopulation marks the cache as loading and starts buffering concurrent
// cache mutations until the population pass completes.
func (s *graphCacheState) beginPopulation() {
        s.updateMtx.Lock()
        defer s.updateMtx.Unlock()

        s.loading = true
        s.pendingUpdates = nil
}

// finishPopulation replays any buffered mutations and marks the cache as ready
// when the initial population completed successfully. If population failed,
// buffered mutations are discarded since the cache won't be used for reads.
func (s *graphCacheState) finishPopulation(loaded bool) {
        s.updateMtx.Lock()
        defer s.updateMtx.Unlock()

        if loaded {
                for _, update := range s.pendingUpdates {
                        update(s.graphCache)
                }

                s.loaded.Store(true)
        } else {
                s.failed.Store(true)
        }

        s.pendingUpdates = nil
        s.loading = false
}

// applyUpdate applies a cache mutation immediately or buffers it when the
// cache is still being populated.
func (s *graphCacheState) applyUpdate(update func(cache *GraphCache)) {
        s.updateMtx.Lock()
        defer s.updateMtx.Unlock()

        if s.loading {
                s.pendingUpdates = append(s.pendingUpdates, update)

                if len(s.pendingUpdates)%pendingUpdatesWarnThreshold == 0 {
                        log.Warnf("Graph cache has %d pending updates "+
                                "buffered during population",
                                len(s.pendingUpdates))
                }

                return
        }

        update(s.graphCache)
}

1	package graphdb
2
3	import (
4	"sync"
5	"sync/atomic"
6	)
7
8	// pendingUpdatesWarnThreshold is the number of buffered cache mutations at
9	// which a warning is logged. A large buffer indicates that cache population is
10	// taking a long time relative to the incoming gossip rate.
11	const pendingUpdatesWarnThreshold = 10_000
12
13	// graphCacheState tracks the in-memory graph cache together with its
14	// population state. The underlying GraphCache is independently thread-safe, so
15	// once reads are allowed to use it, they do not need to hold updateMtx.
16	type graphCacheState struct {
17	graphCache *GraphCache
18	loaded atomic.Bool
19	failed atomic.Bool
20
21	updateMtx sync.Mutex
22	loading bool
23
24	pendingUpdates []func(*GraphCache)
25	}
26
27	// newGraphCacheState constructs a graph cache state with a new cache instance.
28	func newGraphCacheState(preAllocNumNodes int) *graphCacheState {	166✔
29	return &graphCacheState{	166✔
30	graphCache: NewGraphCache(preAllocNumNodes),	166✔
31	}	166✔
32	}	166✔
33
34	// isLoaded reports whether the cache has finished its initial population and
35	// is safe to serve reads from.
36	func (s *graphCacheState) isLoaded() bool {	1,338✔
37	return s.loaded.Load()	1,338✔
38	}	1,338✔
39
40	// isFailed reports whether the cache population attempt has failed.
41	func (s *graphCacheState) isFailed() bool {	4✔
42	return s.failed.Load()	4✔
43	}	4✔
44
45	// stats returns the cache stats if the cache has finished its initial
46	// population.
47	func (s *graphCacheState) stats() (string, bool) {	250✔
48	if !s.isLoaded() {	254✔
49	return "", false	4✔
50	}	4✔
51
52	return s.graphCache.Stats(), true	250✔
53	}
54
55	// beginPopulation marks the cache as loading and starts buffering concurrent
56	// cache mutations until the population pass completes.
57	func (s *graphCacheState) beginPopulation() {	166✔
58	s.updateMtx.Lock()	166✔
59	defer s.updateMtx.Unlock()	166✔
60		166✔
61	s.loading = true	166✔
62	s.pendingUpdates = nil	166✔
63	}	166✔
64
65	// finishPopulation replays any buffered mutations and marks the cache as ready
66	// when the initial population completed successfully. If population failed,
67	// buffered mutations are discarded since the cache won't be used for reads.
68	func (s *graphCacheState) finishPopulation(loaded bool) {	166✔
69	s.updateMtx.Lock()	166✔
70	defer s.updateMtx.Unlock()	166✔
71		166✔
72	if loaded {	329✔
73	for _, update := range s.pendingUpdates {	168✔
74	update(s.graphCache)	5✔
75	}	5✔
76
77	s.loaded.Store(true)	163✔
78	} else {	3✔
79	s.failed.Store(true)	3✔
80	}	3✔
81
82	s.pendingUpdates = nil	166✔
83	s.loading = false	166✔
84	}
85
86	// applyUpdate applies a cache mutation immediately or buffers it when the
87	// cache is still being populated.
88	func (s graphCacheState) applyUpdate(update func(cache GraphCache)) {	6,051✔
89	s.updateMtx.Lock()	6,051✔
90	defer s.updateMtx.Unlock()	6,051✔
91		6,051✔
92	if s.loading {	6,056✔
93	s.pendingUpdates = append(s.pendingUpdates, update)	5✔
94		5✔
95	if len(s.pendingUpdates)%pendingUpdatesWarnThreshold == 0 {	5✔
NEW 96	log.Warnf("Graph cache has %d pending updates "+	×
NEW 97	"buffered during population",	×
NEW 98	len(s.pendingUpdates))	×
NEW 99	}	×
100
101	return	5✔
102	}
103
104	update(s.graphCache)	6,050✔
105	}

lightningnetwork / lnd / 23478841560

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