24453602233

Committed 15 Apr 2026 12:08PM UTC coverage: 51.23% (-11.0%) from 62.198%

Build # 24453602233

Build Type

Pull #10732

github

Committed by

web-flow

Commit Message

Merge d307aaaa7 into 3fc674161

Pull Request Pull Request #10732: channeldb: split channel close into two phases to avoid long DB write-locks

Coverage Stats

201 of 321 new or added lines in 4 files covered. (62.62%)

24474 existing lines in 293 files now uncovered.

115853 of 226144 relevant lines covered (51.23%)

19478.77 hits per line

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

88.24

/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 {	168✔
29	return &graphCacheState{	168✔
30	graphCache: NewGraphCache(preAllocNumNodes),	168✔
31	}	168✔
32	}	168✔
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,329✔
37	return s.loaded.Load()	1,329✔
38	}	1,329✔
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) {	241✔
48	if !s.isLoaded() {	241✔
UNCOV 49	return "", false	×
UNCOV 50	}	×
51
52	return s.graphCache.Stats(), true	241✔
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() {	168✔
58	s.updateMtx.Lock()	168✔
59	defer s.updateMtx.Unlock()	168✔
60		168✔
61	s.loading = true	168✔
62	s.pendingUpdates = nil	168✔
63	}	168✔
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) {	168✔
69	s.updateMtx.Lock()	168✔
70	defer s.updateMtx.Unlock()	168✔
71		168✔
72	if loaded {	333✔
73	for _, update := range s.pendingUpdates {	166✔
74	update(s.graphCache)	1✔
75	}	1✔
76
77	s.loaded.Store(true)	165✔
78	} else {	3✔
79	s.failed.Store(true)	3✔
80	}	3✔
81
82	s.pendingUpdates = nil	168✔
83	s.loading = false	168✔
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,063✔
89	s.updateMtx.Lock()	6,063✔
90	defer s.updateMtx.Unlock()	6,063✔
91		6,063✔
92	if s.loading {	6,064✔
93	s.pendingUpdates = append(s.pendingUpdates, update)	1✔
94		1✔
95	if len(s.pendingUpdates)%pendingUpdatesWarnThreshold == 0 {	1✔
96	log.Warnf("Graph cache has %d pending updates "+	×
97	"buffered during population",	×
98	len(s.pendingUpdates))	×
99	}	×
100
101	return	1✔
102	}
103
104	update(s.graphCache)	6,062✔
105	}

lightningnetwork / lnd / 24453602233

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