19155841408

Committed 07 Nov 2025 02:03AM UTC coverage: 66.675% (-0.04%) from 66.712%

Build # 19155841408

Build Type

Pull #10352

github

Committed by

web-flow

Commit Message

Merge e4313eba8 into 096ab65b1

Pull Request Pull Request #10352: [WIP] chainrpc: return Unavailable while notifier starts

Run Details

137328 of 205965 relevant lines covered (66.68%)

21333.36 hits per line

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

96.36

/autopilot/betweenness_centrality.go

package autopilot

import (
        "context"
        "fmt"
        "sync"
)

// stack is a simple int stack to help with readability of Brandes'
// betweenness centrality implementation below.
type stack struct {
        stack []int
}

func (s *stack) push(v int) {
        s.stack = append(s.stack, v)
}

func (s *stack) top() int {
        return s.stack[len(s.stack)-1]
}

func (s *stack) pop() {
        s.stack = s.stack[:len(s.stack)-1]
}

func (s *stack) empty() bool {
        return len(s.stack) == 0
}

// queue is a simple int queue to help with readability of Brandes'
// betweenness centrality implementation below.
type queue struct {
        queue []int
}

func (q *queue) push(v int) {
        q.queue = append(q.queue, v)
}

func (q *queue) front() int {
        return q.queue[0]
}

func (q *queue) pop() {
        q.queue = q.queue[1:]
}

func (q *queue) empty() bool {
        return len(q.queue) == 0
}

// BetweennessCentrality is a NodeMetric that calculates node betweenness
// centrality using Brandes' algorithm. Betweenness centrality for each node
// is the number of shortest paths passing through that node, not counting
// shortest paths starting or ending at that node. This is a useful metric
// to measure control of individual nodes over the whole network.
type BetweennessCentrality struct {
        // workers number of goroutines are used to parallelize
        // centrality calculation.
        workers int

        // centrality stores original (not normalized) centrality values for
        // each node in the graph.
        centrality map[NodeID]float64

        // min is the minimum centrality in the graph.
        min float64

        // max is the maximum centrality in the graph.
        max float64
}

// NewBetweennessCentralityMetric creates a new BetweennessCentrality instance.
// Users can specify the number of workers to use for calculating centrality.
func NewBetweennessCentralityMetric(workers int) (*BetweennessCentrality, error) {
        // There should be at least one worker.
        if workers < 1 {
                return nil, fmt.Errorf("workers must be positive")
        }
        return &BetweennessCentrality{
                workers: workers,
        }, nil
}

// Name returns the name of the metric.
func (bc *BetweennessCentrality) Name() string {
        return "betweenness_centrality"
}

// betweennessCentrality is the core of Brandes' algorithm.
// We first calculate the shortest paths from the start node s to all other
// nodes with BFS, then update the betweenness centrality values by using
// Brandes' dependency trick.
// For detailed explanation please read:
// https://www.cl.cam.ac.uk/teaching/1617/MLRD/handbook/brandes.html
func betweennessCentrality(g *SimpleGraph, s int, centrality []float64) {
        // pred[w] is the list of nodes that immediately precede w on a
        // shortest path from s to t for each node t.
        pred := make([][]int, len(g.Nodes))

        // sigma[t] is the number of shortest paths between nodes s and t
        // for each node t.
        sigma := make([]int, len(g.Nodes))
        sigma[s] = 1

        // dist[t] holds the distance between s and t for each node t.
        // We initialize this to -1 (meaning infinity) for each t != s.
        dist := make([]int, len(g.Nodes))
        for i := range dist {
                dist[i] = -1
        }

        dist[s] = 0

        var (
                st stack
                q  queue
        )
        q.push(s)

        // BFS to calculate the shortest paths (sigma and pred)
        // from s to t for each node t.
        for !q.empty() {
                v := q.front()
                q.pop()
                st.push(v)

                for _, w := range g.Adj[v] {
                        // If distance from s to w is infinity (-1)
                        // then set it and enqueue w.
                        if dist[w] < 0 {
                                dist[w] = dist[v] + 1
                                q.push(w)
                        }

                        // If w is on a shortest path the update
                        // sigma and add v to w's predecessor list.
                        if dist[w] == dist[v]+1 {
                                sigma[w] += sigma[v]
                                pred[w] = append(pred[w], v)
                        }
                }
        }

        // delta[v] is the ratio of the shortest paths between s and t that go
        // through v and the total number of shortest paths between s and t.
        // If we have delta then the betweenness centrality is simply the sum
        // of delta[w] for each w != s.
        delta := make([]float64, len(g.Nodes))

        for !st.empty() {
                w := st.top()
                st.pop()

                // pred[w] is the list of nodes that immediately precede w on a
                // shortest path from s.
                for _, v := range pred[w] {
                        // Update delta using Brandes' equation.
                        delta[v] += (float64(sigma[v]) / float64(sigma[w])) * (1.0 + delta[w])
                }

                if w != s {
                        // As noted above centrality is simply the sum
                        // of delta[w] for each w != s.
                        centrality[w] += delta[w]
                }
        }
}

// Refresh recalculates and stores centrality values.
func (bc *BetweennessCentrality) Refresh(ctx context.Context,
        graph ChannelGraph) error {

        cache, err := NewSimpleGraph(ctx, graph)
        if err != nil {
                return err
        }

        var wg sync.WaitGroup
        work := make(chan int)
        partials := make(chan []float64, bc.workers)

        // Each worker will compute a partial result.
        // This partial result is a sum of centrality updates
        // on roughly N / workers nodes.
        worker := func() {
                defer wg.Done()
                partial := make([]float64, len(cache.Nodes))

                // Consume the next node, update centrality
                // parital to avoid unnecessary synchronization.
                for node := range work {
                        betweennessCentrality(cache, node, partial)
                }
                partials <- partial
        }

        // Now start the N workers.
        wg.Add(bc.workers)
        for i := 0; i < bc.workers; i++ {
                go worker()
        }

        // Distribute work amongst workers.
        // Should be fair when the graph is sufficiently large.
        for node := range cache.Nodes {
                work <- node
        }

        close(work)
        wg.Wait()
        close(partials)

        // Collect and sum partials for final result.
        centrality := make([]float64, len(cache.Nodes))
        for partial := range partials {
                for i := 0; i < len(partial); i++ {
                        centrality[i] += partial[i]
                }
        }

        // Get min/max to be able to normalize
        // centrality values between 0 and 1.
        bc.min = 0
        bc.max = 0
        if len(centrality) > 0 {
                for _, v := range centrality {
                        if v < bc.min {
                                bc.min = v
                        } else if v > bc.max {
                                bc.max = v
                        }
                }
        }

        // Divide by two as this is an undirected graph.
        bc.min /= 2.0
        bc.max /= 2.0

        bc.centrality = make(map[NodeID]float64)
        for u, value := range centrality {
                // Divide by two as this is an undirected graph.
                bc.centrality[cache.Nodes[u]] = value / 2.0
        }

        return nil
}

// GetMetric returns the current centrality values for each node indexed
// by node id.
func (bc *BetweennessCentrality) GetMetric(normalize bool) map[NodeID]float64 {
        // Normalization factor.
        var z float64
        if (bc.max - bc.min) > 0 {
                z = 1.0 / (bc.max - bc.min)
        }

        centrality := make(map[NodeID]float64)
        for k, v := range bc.centrality {
                if normalize {
                        v = (v - bc.min) * z
                }
                centrality[k] = v
        }

        return centrality
}

1	package autopilot
2
3	import (
4	"context"
5	"fmt"
6	"sync"
7	)
8
9	// stack is a simple int stack to help with readability of Brandes'
10	// betweenness centrality implementation below.
11	type stack struct {
12	stack []int
13	}
14
15	func (s *stack) push(v int) {	16,848✔
16	s.stack = append(s.stack, v)	16,848✔
17	}	16,848✔
18
19	func (s *stack) top() int {	16,848✔
20	return s.stack[len(s.stack)-1]	16,848✔
21	}	16,848✔
22
23	func (s *stack) pop() {	16,848✔
24	s.stack = s.stack[:len(s.stack)-1]	16,848✔
25	}	16,848✔
26
27	func (s *stack) empty() bool {	18,720✔
28	return len(s.stack) == 0	18,720✔
29	}	18,720✔
30
31	// queue is a simple int queue to help with readability of Brandes'
32	// betweenness centrality implementation below.
33	type queue struct {
34	queue []int
35	}
36
37	func (q *queue) push(v int) {	16,848✔
38	q.queue = append(q.queue, v)	16,848✔
39	}	16,848✔
40
41	func (q *queue) front() int {	16,848✔
42	return q.queue[0]	16,848✔
43	}	16,848✔
44
45	func (q *queue) pop() {	16,848✔
46	q.queue = q.queue[1:]	16,848✔
47	}	16,848✔
48
49	func (q *queue) empty() bool {	18,720✔
50	return len(q.queue) == 0	18,720✔
51	}	18,720✔
52
53	// BetweennessCentrality is a NodeMetric that calculates node betweenness
54	// centrality using Brandes' algorithm. Betweenness centrality for each node
55	// is the number of shortest paths passing through that node, not counting
56	// shortest paths starting or ending at that node. This is a useful metric
57	// to measure control of individual nodes over the whole network.
58	type BetweennessCentrality struct {
59	// workers number of goroutines are used to parallelize
60	// centrality calculation.
61	workers int
62
63	// centrality stores original (not normalized) centrality values for
64	// each node in the graph.
65	centrality map[NodeID]float64
66
67	// min is the minimum centrality in the graph.
68	min float64
69
70	// max is the maximum centrality in the graph.
71	max float64
72	}
73
74	// NewBetweennessCentralityMetric creates a new BetweennessCentrality instance.
75	// Users can specify the number of workers to use for calculating centrality.
76	func NewBetweennessCentralityMetric(workers int) (*BetweennessCentrality, error) {	47✔
77	// There should be at least one worker.	47✔
78	if workers < 1 {	49✔
79	return nil, fmt.Errorf("workers must be positive")	2✔
80	}	2✔
81	return &BetweennessCentrality{	45✔
82	workers: workers,	45✔
83	}, nil	45✔
84	}
85
86	// Name returns the name of the metric.
87	func (bc *BetweennessCentrality) Name() string {	×
88	return "betweenness_centrality"	×
89	}	×
90
91	// betweennessCentrality is the core of Brandes' algorithm.
92	// We first calculate the shortest paths from the start node s to all other
93	// nodes with BFS, then update the betweenness centrality values by using
94	// Brandes' dependency trick.
95	// For detailed explanation please read:
96	// https://www.cl.cam.ac.uk/teaching/1617/MLRD/handbook/brandes.html
97	func betweennessCentrality(g *SimpleGraph, s int, centrality []float64) {	1,872✔
98	// pred[w] is the list of nodes that immediately precede w on a	1,872✔
99	// shortest path from s to t for each node t.	1,872✔
100	pred := make([][]int, len(g.Nodes))	1,872✔
101		1,872✔
102	// sigma[t] is the number of shortest paths between nodes s and t	1,872✔
103	// for each node t.	1,872✔
104	sigma := make([]int, len(g.Nodes))	1,872✔
105	sigma[s] = 1	1,872✔
106		1,872✔
107	// dist[t] holds the distance between s and t for each node t.	1,872✔
108	// We initialize this to -1 (meaning infinity) for each t != s.	1,872✔
109	dist := make([]int, len(g.Nodes))	1,872✔
110	for i := range dist {	18,720✔
111	dist[i] = -1	16,848✔
112	}	16,848✔
113
114	dist[s] = 0	1,872✔
115		1,872✔
116	var (	1,872✔
117	st stack	1,872✔
118	q queue	1,872✔
119	)	1,872✔
120	q.push(s)	1,872✔
121		1,872✔
122	// BFS to calculate the shortest paths (sigma and pred)	1,872✔
123	// from s to t for each node t.	1,872✔
124	for !q.empty() {	18,720✔
125	v := q.front()	16,848✔
126	q.pop()	16,848✔
127	st.push(v)	16,848✔
128		16,848✔
129	for _, w := range g.Adj[v] {	69,264✔
130	// If distance from s to w is infinity (-1)	52,416✔
131	// then set it and enqueue w.	52,416✔
132	if dist[w] < 0 {	67,392✔
133	dist[w] = dist[v] + 1	14,976✔
134	q.push(w)	14,976✔
135	}	14,976✔
136
137	// If w is on a shortest path the update
138	// sigma and add v to w's predecessor list.
139	if dist[w] == dist[v]+1 {	71,136✔
140	sigma[w] += sigma[v]	18,720✔
141	pred[w] = append(pred[w], v)	18,720✔
142	}	18,720✔
143	}
144	}
145
146	// delta[v] is the ratio of the shortest paths between s and t that go
147	// through v and the total number of shortest paths between s and t.
148	// If we have delta then the betweenness centrality is simply the sum
149	// of delta[w] for each w != s.
150	delta := make([]float64, len(g.Nodes))	1,872✔
151		1,872✔
152	for !st.empty() {	18,720✔
153	w := st.top()	16,848✔
154	st.pop()	16,848✔
155		16,848✔
156	// pred[w] is the list of nodes that immediately precede w on a	16,848✔
157	// shortest path from s.	16,848✔
158	for _, v := range pred[w] {	35,568✔
159	// Update delta using Brandes' equation.	18,720✔
160	delta[v] += (float64(sigma[v]) / float64(sigma[w])) * (1.0 + delta[w])	18,720✔
161	}	18,720✔
162
163	if w != s {	31,824✔
164	// As noted above centrality is simply the sum	14,976✔
165	// of delta[w] for each w != s.	14,976✔
166	centrality[w] += delta[w]	14,976✔
167	}	14,976✔
168	}
169	}
170
171	// Refresh recalculates and stores centrality values.
172	func (bc *BetweennessCentrality) Refresh(ctx context.Context,
173	graph ChannelGraph) error {	210✔
174		210✔
175	cache, err := NewSimpleGraph(ctx, graph)	210✔
176	if err != nil {	210✔
177	return err	×
178	}	×
179
180	var wg sync.WaitGroup	210✔
181	work := make(chan int)	210✔
182	partials := make(chan []float64, bc.workers)	210✔
183		210✔
184	// Each worker will compute a partial result.	210✔
185	// This partial result is a sum of centrality updates	210✔
186	// on roughly N / workers nodes.	210✔
187	worker := func() {	1,238✔
188	defer wg.Done()	1,028✔
189	partial := make([]float64, len(cache.Nodes))	1,028✔
190		1,028✔
191	// Consume the next node, update centrality	1,028✔
192	// parital to avoid unnecessary synchronization.	1,028✔
193	for node := range work {	2,900✔
194	betweennessCentrality(cache, node, partial)	1,872✔
195	}	1,872✔
196	partials <- partial	1,028✔
197	}
198
199	// Now start the N workers.
200	wg.Add(bc.workers)	210✔
201	for i := 0; i < bc.workers; i++ {	1,238✔
202	go worker()	1,028✔
203	}	1,028✔
204
205	// Distribute work amongst workers.
206	// Should be fair when the graph is sufficiently large.
207	for node := range cache.Nodes {	2,082✔
208	work <- node	1,872✔
209	}	1,872✔
210
211	close(work)	210✔
212	wg.Wait()	210✔
213	close(partials)	210✔
214		210✔
215	// Collect and sum partials for final result.	210✔
216	centrality := make([]float64, len(cache.Nodes))	210✔
217	for partial := range partials {	1,238✔
218	for i := 0; i < len(partial); i++ {	10,262✔
219	centrality[i] += partial[i]	9,234✔
220	}	9,234✔
221	}
222
223	// Get min/max to be able to normalize
224	// centrality values between 0 and 1.
225	bc.min = 0	210✔
226	bc.max = 0	210✔
227	if len(centrality) > 0 {	418✔
228	for _, v := range centrality {	2,080✔
229	if v < bc.min {	1,872✔
230	bc.min = v	×
231	} else if v > bc.max {	2,302✔
232	bc.max = v	430✔
233	}	430✔
234	}
235	}
236
237	// Divide by two as this is an undirected graph.
238	bc.min /= 2.0	210✔
239	bc.max /= 2.0	210✔
240		210✔
241	bc.centrality = make(map[NodeID]float64)	210✔
242	for u, value := range centrality {	2,082✔
243	// Divide by two as this is an undirected graph.	1,872✔
244	bc.centrality[cache.Nodes[u]] = value / 2.0	1,872✔
245	}	1,872✔
246
247	return nil	210✔
248	}
249
250	// GetMetric returns the current centrality values for each node indexed
251	// by node id.
252	func (bc *BetweennessCentrality) GetMetric(normalize bool) map[NodeID]float64 {	220✔
253	// Normalization factor.	220✔
254	var z float64	220✔
255	if (bc.max - bc.min) > 0 {	436✔
256	z = 1.0 / (bc.max - bc.min)	216✔
257	}	216✔
258
259	centrality := make(map[NodeID]float64)	220✔
260	for k, v := range bc.centrality {	2,164✔
261	if normalize {	3,816✔
262	v = (v - bc.min) * z	1,872✔
263	}	1,872✔
264	centrality[k] = v	1,944✔
265	}
266
267	return centrality	220✔
268	}

lightningnetwork / lnd / 19155841408

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