15747152963

Committed 19 Jun 2025 01:22AM UTC coverage: 58.151% (-10.1%) from 68.248%

Build # 15747152963

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #9528 from Roasbeef/res-opt

fn: implement ResultOpt type for operations with optional values

Run Details

97778 of 168145 relevant lines covered (58.15%)

1.81 hits per line

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

3.64

/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	s.stack = append(s.stack, v)	×
17	}	×
18
19	func (s *stack) top() int {	×
20	return s.stack[len(s.stack)-1]	×
21	}	×
22
23	func (s *stack) pop() {	×
24	s.stack = s.stack[:len(s.stack)-1]	×
25	}	×
26
27	func (s *stack) empty() bool {	×
28	return len(s.stack) == 0	×
29	}	×
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) {	×
38	q.queue = append(q.queue, v)	×
39	}	×
40
41	func (q *queue) front() int {	×
42	return q.queue[0]	×
43	}	×
44
45	func (q *queue) pop() {	×
46	q.queue = q.queue[1:]	×
47	}	×
48
49	func (q *queue) empty() bool {	×
50	return len(q.queue) == 0	×
51	}	×
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) {	3✔
77	// There should be at least one worker.	3✔
78	if workers < 1 {	3✔
79	return nil, fmt.Errorf("workers must be positive")	×
80	}	×
81	return &BetweennessCentrality{	3✔
82	workers: workers,	3✔
83	}, nil	3✔
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) {	×
98	// pred[w] is the list of nodes that immediately precede w on a	×
99	// shortest path from s to t for each node t.	×
100	pred := make([][]int, len(g.Nodes))	×
101		×
102	// sigma[t] is the number of shortest paths between nodes s and t	×
103	// for each node t.	×
104	sigma := make([]int, len(g.Nodes))	×
105	sigma[s] = 1	×
106		×
107	// dist[t] holds the distance between s and t for each node t.	×
108	// We initialize this to -1 (meaning infinity) for each t != s.	×
109	dist := make([]int, len(g.Nodes))	×
110	for i := range dist {	×
111	dist[i] = -1	×
112	}	×
113
114	dist[s] = 0	×
115		×
116	var (	×
117	st stack	×
118	q queue	×
119	)	×
120	q.push(s)	×
121		×
122	// BFS to calculate the shortest paths (sigma and pred)	×
123	// from s to t for each node t.	×
124	for !q.empty() {	×
125	v := q.front()	×
126	q.pop()	×
127	st.push(v)	×
128		×
129	for _, w := range g.Adj[v] {	×
130	// If distance from s to w is infinity (-1)	×
131	// then set it and enqueue w.	×
132	if dist[w] < 0 {	×
133	dist[w] = dist[v] + 1	×
134	q.push(w)	×
135	}	×
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 {	×
140	sigma[w] += sigma[v]	×
141	pred[w] = append(pred[w], v)	×
142	}	×
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))	×
151		×
152	for !st.empty() {	×
153	w := st.top()	×
154	st.pop()	×
155		×
156	// pred[w] is the list of nodes that immediately precede w on a	×
157	// shortest path from s.	×
158	for _, v := range pred[w] {	×
159	// Update delta using Brandes' equation.	×
160	delta[v] += (float64(sigma[v]) / float64(sigma[w])) * (1.0 + delta[w])	×
161	}	×
162
163	if w != s {	×
164	// As noted above centrality is simply the sum	×
165	// of delta[w] for each w != s.	×
166	centrality[w] += delta[w]	×
167	}	×
168	}
169	}
170
171	// Refresh recalculates and stores centrality values.
172	func (bc *BetweennessCentrality) Refresh(ctx context.Context,
173	graph ChannelGraph) error {	×
174		×
175	cache, err := NewSimpleGraph(ctx, graph)	×
176	if err != nil {	×
177	return err	×
178	}	×
179
180	var wg sync.WaitGroup	×
181	work := make(chan int)	×
182	partials := make(chan []float64, bc.workers)	×
183		×
184	// Each worker will compute a partial result.	×
185	// This partial result is a sum of centrality updates	×
186	// on roughly N / workers nodes.	×
187	worker := func() {	×
188	defer wg.Done()	×
189	partial := make([]float64, len(cache.Nodes))	×
190		×
191	// Consume the next node, update centrality	×
192	// parital to avoid unnecessary synchronization.	×
193	for node := range work {	×
194	betweennessCentrality(cache, node, partial)	×
195	}	×
196	partials <- partial	×
197	}
198
199	// Now start the N workers.
200	wg.Add(bc.workers)	×
201	for i := 0; i < bc.workers; i++ {	×
202	go worker()	×
203	}	×
204
205	// Distribute work amongst workers.
206	// Should be fair when the graph is sufficiently large.
207	for node := range cache.Nodes {	×
208	work <- node	×
209	}	×
210
211	close(work)	×
212	wg.Wait()	×
213	close(partials)	×
214		×
215	// Collect and sum partials for final result.	×
216	centrality := make([]float64, len(cache.Nodes))	×
217	for partial := range partials {	×
218	for i := 0; i < len(partial); i++ {	×
219	centrality[i] += partial[i]	×
220	}	×
221	}
222
223	// Get min/max to be able to normalize
224	// centrality values between 0 and 1.
225	bc.min = 0	×
226	bc.max = 0	×
227	if len(centrality) > 0 {	×
228	for _, v := range centrality {	×
229	if v < bc.min {	×
230	bc.min = v	×
231	} else if v > bc.max {	×
232	bc.max = v	×
233	}	×
234	}
235	}
236
237	// Divide by two as this is an undirected graph.
238	bc.min /= 2.0	×
239	bc.max /= 2.0	×
240		×
241	bc.centrality = make(map[NodeID]float64)	×
242	for u, value := range centrality {	×
243	// Divide by two as this is an undirected graph.	×
244	bc.centrality[cache.Nodes[u]] = value / 2.0	×
245	}	×
246
247	return nil	×
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 {	×
253	// Normalization factor.	×
254	var z float64	×
255	if (bc.max - bc.min) > 0 {	×
256	z = 1.0 / (bc.max - bc.min)	×
257	}	×
258
259	centrality := make(map[NodeID]float64)	×
260	for k, v := range bc.centrality {	×
261	if normalize {	×
262	v = (v - bc.min) * z	×
263	}	×
264	centrality[k] = v	×
265	}
266
267	return centrality	×
268	}

lightningnetwork / lnd / 15747152963

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