-
Notifications
You must be signed in to change notification settings - Fork 977
/
node_set.go
120 lines (105 loc) · 3.54 KB
/
node_set.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
package eth1
import (
"context"
"github.com/pkg/errors"
"github.com/prysmaticlabs/prysm/v5/testing/endtoend/helpers"
e2e "github.com/prysmaticlabs/prysm/v5/testing/endtoend/params"
e2etypes "github.com/prysmaticlabs/prysm/v5/testing/endtoend/types"
)
// NodeSet represents a set of Eth1 nodes, none of which is a mining node.
type NodeSet struct {
e2etypes.ComponentRunner
started chan struct{}
enr string
nodes []e2etypes.ComponentRunner
}
// NewNodeSet creates and returns a set of Eth1 nodes.
func NewNodeSet() *NodeSet {
return &NodeSet{
started: make(chan struct{}, 1),
}
}
// SetMinerENR sets the miner's enode, used to connect to the miner through P2P.
func (s *NodeSet) SetMinerENR(enr string) {
s.enr = enr
}
// Start starts all the execution nodes in set.
func (s *NodeSet) Start(ctx context.Context) error {
// Create Eth1 nodes. The number of nodes is the same as the number of beacon nodes.
// We want each beacon node to connect to its own Eth1 node.
// We start up one Eth1 node less than the beacon node count because the first
// beacon node will connect to the already existing Eth1 miner.
totalNodeCount := e2e.TestParams.BeaconNodeCount + e2e.TestParams.LighthouseBeaconNodeCount - 1
nodes := make([]e2etypes.ComponentRunner, totalNodeCount)
for i := 0; i < totalNodeCount; i++ {
// We start indexing nodes from 1 because the miner has an implicit 0 index.
node := NewNode(i+1, s.enr)
nodes[i] = node
}
s.nodes = nodes
// Wait for all nodes to finish their job (blocking).
// Once nodes are ready passed in handler function will be called.
return helpers.WaitOnNodes(ctx, nodes, func() {
// All nodes started, close channel, so that all services waiting on a set, can proceed.
close(s.started)
})
}
// Started checks whether execution node set is started and all nodes are ready to be queried.
func (s *NodeSet) Started() <-chan struct{} {
return s.started
}
// Pause pauses the component and its underlying process.
func (s *NodeSet) Pause() error {
for _, n := range s.nodes {
if err := n.Pause(); err != nil {
return err
}
}
return nil
}
// Resume resumes the component and its underlying process.
func (s *NodeSet) Resume() error {
for _, n := range s.nodes {
if err := n.Resume(); err != nil {
return err
}
}
return nil
}
// Stop stops the component and its underlying process.
func (s *NodeSet) Stop() error {
for _, n := range s.nodes {
if err := n.Stop(); err != nil {
return err
}
}
return nil
}
// PauseAtIndex pauses the component and its underlying process at the desired index.
func (s *NodeSet) PauseAtIndex(i int) error {
if i >= len(s.nodes) {
return errors.Errorf("provided index exceeds slice size: %d >= %d", i, len(s.nodes))
}
return s.nodes[i].Pause()
}
// ResumeAtIndex resumes the component and its underlying process at the desired index.
func (s *NodeSet) ResumeAtIndex(i int) error {
if i >= len(s.nodes) {
return errors.Errorf("provided index exceeds slice size: %d >= %d", i, len(s.nodes))
}
return s.nodes[i].Resume()
}
// StopAtIndex stops the component and its underlying process at the desired index.
func (s *NodeSet) StopAtIndex(i int) error {
if i >= len(s.nodes) {
return errors.Errorf("provided index exceeds slice size: %d >= %d", i, len(s.nodes))
}
return s.nodes[i].Stop()
}
// ComponentAtIndex returns the component at the provided index.
func (s *NodeSet) ComponentAtIndex(i int) (e2etypes.ComponentRunner, error) {
if i >= len(s.nodes) {
return nil, errors.Errorf("provided index exceeds slice size: %d >= %d", i, len(s.nodes))
}
return s.nodes[i], nil
}