14193549836

Committed 01 Apr 2025 10:40AM UTC coverage: 69.046% (+0.007%) from 69.039%

Build # 14193549836

Build Type

Pull #9665

github

Committed by

web-flow

Commit Message

Merge e8825f209 into b01f4e514

Pull Request Pull Request #9665: kvdb: bump etcd libs to v3.5.12

Run Details

133439 of 193262 relevant lines covered (69.05%)

22119.45 hits per line

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

96.34

/autopilot/betweenness_centrality.go

package autopilot

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

lightningnetwork / lnd / 14193549836

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