p2p, swarm: fix node up races by granular locking (#18976)

* swarm/network: DRY out repeated giga comment

I not necessarily agree with the way we wait for event propagation.
But I truly disagree with having duplicated giga comments.

* p2p/simulations: encapsulate Node.Up field so we avoid data races

The Node.Up field was accessed concurrently without "proper" locking.
There was a lock on Network and that was used sometimes to access
the  field. Other times the locking was missed and we had
a data race.

For example: https://github.com/ethereum/go-ethereum/pull/18464
The case above was solved, but there were still intermittent/hard to
reproduce races. So let's solve the issue permanently.

resolves: ethersphere/go-ethereum#1146

* p2p/simulations: fix unmarshal of simulations.Node

Making Node.Up field private in 13292ee897e345045fbfab3bda23a77589a271c1
broke TestHTTPNetwork and TestHTTPSnapshot. Because the default
UnmarshalJSON does not handle unexported fields.

Important: The fix is partial and not proper to my taste. But I cut
scope as I think the fix may require a change to the current
serialization format. New ticket:
https://github.com/ethersphere/go-ethereum/issues/1177

* p2p/simulations: Add a sanity test case for Node.Config UnmarshalJSON

* p2p/simulations: revert back to defer Unlock() pattern for Network

It's a good patten to call `defer Unlock()` right after `Lock()` so
(new) error cases won't miss to unlock. Let's get back to that pattern.

The patten was abandoned in 85a79b3ad3,
while fixing a data race. That data race does not exist anymore,
since the Node.Up field got hidden behind its own lock.

* p2p/simulations: consistent naming for test providers Node.UnmarshalJSON

* p2p/simulations: remove JSON annotation from private fields of Node

As unexported fields are not serialized.

* p2p/simulations: fix deadlock in Network.GetRandomDownNode()

Problem: GetRandomDownNode() locks -> getDownNodeIDs() ->
GetNodes() tries to lock -> deadlock

On Network type, unexported functions must assume that `net.lock`
is already acquired and should not call exported functions which
might try to lock again.

* p2p/simulations: ensure method conformity for Network

Connect* methods were moved to p2p/simulations.Network from
swarm/network/simulation. However these new methods did not follow
the pattern of Network methods, i.e., all exported method locks
the whole Network either for read or write.

* p2p/simulations: fix deadlock during network shutdown

`TestDiscoveryPersistenceSimulationSimAdapter` often got into deadlock.
The execution was stuck on two locks, i.e, `Kademlia.lock` and
`p2p/simulations.Network.lock`. Usually the test got stuck once in each
20 executions with high confidence.

`Kademlia` was stuck in `Kademlia.EachAddr()` and `Network` in
`Network.Stop()`.

Solution: in `Network.Stop()` `net.lock` must be released before
calling `node.Stop()` as stopping a node (somehow - I did not find
the exact code path) causes `Network.InitConn()` to be called from
`Kademlia.SuggestPeer()` and that blocks on `net.lock`.

Related ticket: https://github.com/ethersphere/go-ethereum/issues/1223

* swarm/state: simplify if statement in DBStore.Put()

* p2p/simulations: remove faulty godoc from private function

The comment started with the wrong method name.

The method is simple and self explanatory. Also, it's private.
=> Let's just remove the comment.
This commit is contained in:
Ferenc Szabo 2019-02-18 07:38:14 +01:00 committed by Viktor Trón
parent 12ca3b172a
commit 50b872bf05
12 changed files with 326 additions and 126 deletions

View File

@ -32,6 +32,9 @@ var (
// It is useful when constructing a chain network topology // It is useful when constructing a chain network topology
// when Network adds and removes nodes dynamically. // when Network adds and removes nodes dynamically.
func (net *Network) ConnectToLastNode(id enode.ID) (err error) { func (net *Network) ConnectToLastNode(id enode.ID) (err error) {
net.lock.Lock()
defer net.lock.Unlock()
ids := net.getUpNodeIDs() ids := net.getUpNodeIDs()
l := len(ids) l := len(ids)
if l < 2 { if l < 2 {
@ -41,29 +44,35 @@ func (net *Network) ConnectToLastNode(id enode.ID) (err error) {
if last == id { if last == id {
last = ids[l-2] last = ids[l-2]
} }
return net.connect(last, id) return net.connectNotConnected(last, id)
} }
// ConnectToRandomNode connects the node with provided NodeID // ConnectToRandomNode connects the node with provided NodeID
// to a random node that is up. // to a random node that is up.
func (net *Network) ConnectToRandomNode(id enode.ID) (err error) { func (net *Network) ConnectToRandomNode(id enode.ID) (err error) {
selected := net.GetRandomUpNode(id) net.lock.Lock()
defer net.lock.Unlock()
selected := net.getRandomUpNode(id)
if selected == nil { if selected == nil {
return ErrNodeNotFound return ErrNodeNotFound
} }
return net.connect(selected.ID(), id) return net.connectNotConnected(selected.ID(), id)
} }
// ConnectNodesFull connects all nodes one to another. // ConnectNodesFull connects all nodes one to another.
// It provides a complete connectivity in the network // It provides a complete connectivity in the network
// which should be rarely needed. // which should be rarely needed.
func (net *Network) ConnectNodesFull(ids []enode.ID) (err error) { func (net *Network) ConnectNodesFull(ids []enode.ID) (err error) {
net.lock.Lock()
defer net.lock.Unlock()
if ids == nil { if ids == nil {
ids = net.getUpNodeIDs() ids = net.getUpNodeIDs()
} }
for i, lid := range ids { for i, lid := range ids {
for _, rid := range ids[i+1:] { for _, rid := range ids[i+1:] {
if err = net.connect(lid, rid); err != nil { if err = net.connectNotConnected(lid, rid); err != nil {
return err return err
} }
} }
@ -74,12 +83,19 @@ func (net *Network) ConnectNodesFull(ids []enode.ID) (err error) {
// ConnectNodesChain connects all nodes in a chain topology. // ConnectNodesChain connects all nodes in a chain topology.
// If ids argument is nil, all nodes that are up will be connected. // If ids argument is nil, all nodes that are up will be connected.
func (net *Network) ConnectNodesChain(ids []enode.ID) (err error) { func (net *Network) ConnectNodesChain(ids []enode.ID) (err error) {
net.lock.Lock()
defer net.lock.Unlock()
return net.connectNodesChain(ids)
}
func (net *Network) connectNodesChain(ids []enode.ID) (err error) {
if ids == nil { if ids == nil {
ids = net.getUpNodeIDs() ids = net.getUpNodeIDs()
} }
l := len(ids) l := len(ids)
for i := 0; i < l-1; i++ { for i := 0; i < l-1; i++ {
if err := net.connect(ids[i], ids[i+1]); err != nil { if err := net.connectNotConnected(ids[i], ids[i+1]); err != nil {
return err return err
} }
} }
@ -89,6 +105,9 @@ func (net *Network) ConnectNodesChain(ids []enode.ID) (err error) {
// ConnectNodesRing connects all nodes in a ring topology. // ConnectNodesRing connects all nodes in a ring topology.
// If ids argument is nil, all nodes that are up will be connected. // If ids argument is nil, all nodes that are up will be connected.
func (net *Network) ConnectNodesRing(ids []enode.ID) (err error) { func (net *Network) ConnectNodesRing(ids []enode.ID) (err error) {
net.lock.Lock()
defer net.lock.Unlock()
if ids == nil { if ids == nil {
ids = net.getUpNodeIDs() ids = net.getUpNodeIDs()
} }
@ -96,15 +115,18 @@ func (net *Network) ConnectNodesRing(ids []enode.ID) (err error) {
if l < 2 { if l < 2 {
return nil return nil
} }
if err := net.ConnectNodesChain(ids); err != nil { if err := net.connectNodesChain(ids); err != nil {
return err return err
} }
return net.connect(ids[l-1], ids[0]) return net.connectNotConnected(ids[l-1], ids[0])
} }
// ConnectNodesStar connects all nodes into a star topology // ConnectNodesStar connects all nodes into a star topology
// If ids argument is nil, all nodes that are up will be connected. // If ids argument is nil, all nodes that are up will be connected.
func (net *Network) ConnectNodesStar(ids []enode.ID, center enode.ID) (err error) { func (net *Network) ConnectNodesStar(ids []enode.ID, center enode.ID) (err error) {
net.lock.Lock()
defer net.lock.Unlock()
if ids == nil { if ids == nil {
ids = net.getUpNodeIDs() ids = net.getUpNodeIDs()
} }
@ -112,16 +134,15 @@ func (net *Network) ConnectNodesStar(ids []enode.ID, center enode.ID) (err error
if center == id { if center == id {
continue continue
} }
if err := net.connect(center, id); err != nil { if err := net.connectNotConnected(center, id); err != nil {
return err return err
} }
} }
return nil return nil
} }
// connect connects two nodes but ignores already connected error. func (net *Network) connectNotConnected(oneID, otherID enode.ID) error {
func (net *Network) connect(oneID, otherID enode.ID) error { return ignoreAlreadyConnectedErr(net.connect(oneID, otherID))
return ignoreAlreadyConnectedErr(net.Connect(oneID, otherID))
} }
func ignoreAlreadyConnectedErr(err error) error { func ignoreAlreadyConnectedErr(err error) error {

View File

@ -100,7 +100,7 @@ func ControlEvent(v interface{}) *Event {
func (e *Event) String() string { func (e *Event) String() string {
switch e.Type { switch e.Type {
case EventTypeNode: case EventTypeNode:
return fmt.Sprintf("<node-event> id: %s up: %t", e.Node.ID().TerminalString(), e.Node.Up) return fmt.Sprintf("<node-event> id: %s up: %t", e.Node.ID().TerminalString(), e.Node.Up())
case EventTypeConn: case EventTypeConn:
return fmt.Sprintf("<conn-event> nodes: %s->%s up: %t", e.Conn.One.TerminalString(), e.Conn.Other.TerminalString(), e.Conn.Up) return fmt.Sprintf("<conn-event> nodes: %s->%s up: %t", e.Conn.One.TerminalString(), e.Conn.Other.TerminalString(), e.Conn.Up)
case EventTypeMsg: case EventTypeMsg:

View File

@ -421,14 +421,15 @@ type expectEvents struct {
} }
func (t *expectEvents) nodeEvent(id string, up bool) *Event { func (t *expectEvents) nodeEvent(id string, up bool) *Event {
node := Node{
Config: &adapters.NodeConfig{
ID: enode.HexID(id),
},
up: up,
}
return &Event{ return &Event{
Type: EventTypeNode, Type: EventTypeNode,
Node: &Node{ Node: &node,
Config: &adapters.NodeConfig{
ID: enode.HexID(id),
},
Up: up,
},
} }
} }
@ -480,6 +481,7 @@ loop:
} }
func (t *expectEvents) expect(events ...*Event) { func (t *expectEvents) expect(events ...*Event) {
t.Helper()
timeout := time.After(10 * time.Second) timeout := time.After(10 * time.Second)
i := 0 i := 0
for { for {
@ -501,8 +503,8 @@ func (t *expectEvents) expect(events ...*Event) {
if event.Node.ID() != expected.Node.ID() { if event.Node.ID() != expected.Node.ID() {
t.Fatalf("expected node event %d to have id %q, got %q", i, expected.Node.ID().TerminalString(), event.Node.ID().TerminalString()) t.Fatalf("expected node event %d to have id %q, got %q", i, expected.Node.ID().TerminalString(), event.Node.ID().TerminalString())
} }
if event.Node.Up != expected.Node.Up { if event.Node.Up() != expected.Node.Up() {
t.Fatalf("expected node event %d to have up=%t, got up=%t", i, expected.Node.Up, event.Node.Up) t.Fatalf("expected node event %d to have up=%t, got up=%t", i, expected.Node.Up(), event.Node.Up())
} }
case EventTypeConn: case EventTypeConn:

View File

@ -90,15 +90,12 @@ func TestMocker(t *testing.T) {
for { for {
select { select {
case event := <-events: case event := <-events:
//if the event is a node Up event only if isNodeUp(event) {
if event.Node != nil && event.Node.Up {
//add the correspondent node ID to the map //add the correspondent node ID to the map
nodemap[event.Node.Config.ID] = true nodemap[event.Node.Config.ID] = true
//this means all nodes got a nodeUp event, so we can continue the test //this means all nodes got a nodeUp event, so we can continue the test
if len(nodemap) == nodeCount { if len(nodemap) == nodeCount {
nodesComplete = true nodesComplete = true
//wait for 3s as the mocker will need time to connect the nodes
//time.Sleep( 3 *time.Second)
} }
} else if event.Conn != nil && nodesComplete { } else if event.Conn != nil && nodesComplete {
connCount += 1 connCount += 1
@ -169,3 +166,7 @@ func TestMocker(t *testing.T) {
t.Fatalf("Expected empty list of nodes, got: %d", len(nodesInfo)) t.Fatalf("Expected empty list of nodes, got: %d", len(nodesInfo))
} }
} }
func isNodeUp(event *Event) bool {
return event.Node != nil && event.Node.Up()
}

View File

@ -137,7 +137,7 @@ func (net *Network) Config() *NetworkConfig {
// StartAll starts all nodes in the network // StartAll starts all nodes in the network
func (net *Network) StartAll() error { func (net *Network) StartAll() error {
for _, node := range net.Nodes { for _, node := range net.Nodes {
if node.Up { if node.Up() {
continue continue
} }
if err := net.Start(node.ID()); err != nil { if err := net.Start(node.ID()); err != nil {
@ -150,7 +150,7 @@ func (net *Network) StartAll() error {
// StopAll stops all nodes in the network // StopAll stops all nodes in the network
func (net *Network) StopAll() error { func (net *Network) StopAll() error {
for _, node := range net.Nodes { for _, node := range net.Nodes {
if !node.Up { if !node.Up() {
continue continue
} }
if err := net.Stop(node.ID()); err != nil { if err := net.Stop(node.ID()); err != nil {
@ -169,27 +169,23 @@ func (net *Network) Start(id enode.ID) error {
// snapshots // snapshots
func (net *Network) startWithSnapshots(id enode.ID, snapshots map[string][]byte) error { func (net *Network) startWithSnapshots(id enode.ID, snapshots map[string][]byte) error {
net.lock.Lock() net.lock.Lock()
defer net.lock.Unlock()
node := net.getNode(id) node := net.getNode(id)
if node == nil { if node == nil {
net.lock.Unlock()
return fmt.Errorf("node %v does not exist", id) return fmt.Errorf("node %v does not exist", id)
} }
if node.Up { if node.Up() {
net.lock.Unlock()
return fmt.Errorf("node %v already up", id) return fmt.Errorf("node %v already up", id)
} }
log.Trace("Starting node", "id", id, "adapter", net.nodeAdapter.Name()) log.Trace("Starting node", "id", id, "adapter", net.nodeAdapter.Name())
if err := node.Start(snapshots); err != nil { if err := node.Start(snapshots); err != nil {
net.lock.Unlock()
log.Warn("Node startup failed", "id", id, "err", err) log.Warn("Node startup failed", "id", id, "err", err)
return err return err
} }
node.Up = true node.SetUp(true)
log.Info("Started node", "id", id) log.Info("Started node", "id", id)
ev := NewEvent(node) ev := NewEvent(node)
net.lock.Unlock()
net.events.Send(ev) net.events.Send(ev)
// subscribe to peer events // subscribe to peer events
@ -220,7 +216,7 @@ func (net *Network) watchPeerEvents(id enode.ID, events chan *p2p.PeerEvent, sub
if node == nil { if node == nil {
return return
} }
node.Up = false node.SetUp(false)
ev := NewEvent(node) ev := NewEvent(node)
net.events.Send(ev) net.events.Send(ev)
}() }()
@ -258,30 +254,42 @@ func (net *Network) watchPeerEvents(id enode.ID, events chan *p2p.PeerEvent, sub
// Stop stops the node with the given ID // Stop stops the node with the given ID
func (net *Network) Stop(id enode.ID) error { func (net *Network) Stop(id enode.ID) error {
net.lock.Lock() // IMPORTANT: node.Stop() must NOT be called under net.lock as
node := net.getNode(id) // node.Reachable() closure has a reference to the network and
if node == nil { // calls net.InitConn() what also locks the network. => DEADLOCK
net.lock.Unlock() // That holds until the following ticket is not resolved:
return fmt.Errorf("node %v does not exist", id)
}
if !node.Up {
net.lock.Unlock()
return fmt.Errorf("node %v already down", id)
}
node.Up = false
net.lock.Unlock()
err := node.Stop() var err error
if err != nil {
node, err := func() (*Node, error) {
net.lock.Lock() net.lock.Lock()
node.Up = true defer net.lock.Unlock()
net.lock.Unlock()
node := net.getNode(id)
if node == nil {
return nil, fmt.Errorf("node %v does not exist", id)
}
if !node.Up() {
return nil, fmt.Errorf("node %v already down", id)
}
node.SetUp(false)
return node, nil
}()
if err != nil {
return err
}
err = node.Stop() // must be called without net.lock
net.lock.Lock()
defer net.lock.Unlock()
if err != nil {
node.SetUp(true)
return err return err
} }
log.Info("Stopped node", "id", id, "err", err) log.Info("Stopped node", "id", id, "err", err)
net.lock.Lock()
ev := ControlEvent(node) ev := ControlEvent(node)
net.lock.Unlock()
net.events.Send(ev) net.events.Send(ev)
return nil return nil
} }
@ -289,8 +297,14 @@ func (net *Network) Stop(id enode.ID) error {
// Connect connects two nodes together by calling the "admin_addPeer" RPC // Connect connects two nodes together by calling the "admin_addPeer" RPC
// method on the "one" node so that it connects to the "other" node // method on the "one" node so that it connects to the "other" node
func (net *Network) Connect(oneID, otherID enode.ID) error { func (net *Network) Connect(oneID, otherID enode.ID) error {
net.lock.Lock()
defer net.lock.Unlock()
return net.connect(oneID, otherID)
}
func (net *Network) connect(oneID, otherID enode.ID) error {
log.Debug("Connecting nodes with addPeer", "id", oneID, "other", otherID) log.Debug("Connecting nodes with addPeer", "id", oneID, "other", otherID)
conn, err := net.InitConn(oneID, otherID) conn, err := net.initConn(oneID, otherID)
if err != nil { if err != nil {
return err return err
} }
@ -388,6 +402,14 @@ func (net *Network) GetNode(id enode.ID) *Node {
return net.getNode(id) return net.getNode(id)
} }
func (net *Network) getNode(id enode.ID) *Node {
i, found := net.nodeMap[id]
if !found {
return nil
}
return net.Nodes[i]
}
// GetNode gets the node with the given name, returning nil if the node does // GetNode gets the node with the given name, returning nil if the node does
// not exist // not exist
func (net *Network) GetNodeByName(name string) *Node { func (net *Network) GetNodeByName(name string) *Node {
@ -410,28 +432,29 @@ func (net *Network) GetNodes() (nodes []*Node) {
net.lock.RLock() net.lock.RLock()
defer net.lock.RUnlock() defer net.lock.RUnlock()
nodes = append(nodes, net.Nodes...) return net.getNodes()
return nodes
} }
func (net *Network) getNode(id enode.ID) *Node { func (net *Network) getNodes() (nodes []*Node) {
i, found := net.nodeMap[id] nodes = append(nodes, net.Nodes...)
if !found { return nodes
return nil
}
return net.Nodes[i]
} }
// GetRandomUpNode returns a random node on the network, which is running. // GetRandomUpNode returns a random node on the network, which is running.
func (net *Network) GetRandomUpNode(excludeIDs ...enode.ID) *Node { func (net *Network) GetRandomUpNode(excludeIDs ...enode.ID) *Node {
net.lock.RLock() net.lock.RLock()
defer net.lock.RUnlock() defer net.lock.RUnlock()
return net.getRandomUpNode(excludeIDs...)
}
// GetRandomUpNode returns a random node on the network, which is running.
func (net *Network) getRandomUpNode(excludeIDs ...enode.ID) *Node {
return net.getRandomNode(net.getUpNodeIDs(), excludeIDs) return net.getRandomNode(net.getUpNodeIDs(), excludeIDs)
} }
func (net *Network) getUpNodeIDs() (ids []enode.ID) { func (net *Network) getUpNodeIDs() (ids []enode.ID) {
for _, node := range net.Nodes { for _, node := range net.Nodes {
if node.Up { if node.Up() {
ids = append(ids, node.ID()) ids = append(ids, node.ID())
} }
} }
@ -446,8 +469,8 @@ func (net *Network) GetRandomDownNode(excludeIDs ...enode.ID) *Node {
} }
func (net *Network) getDownNodeIDs() (ids []enode.ID) { func (net *Network) getDownNodeIDs() (ids []enode.ID) {
for _, node := range net.GetNodes() { for _, node := range net.getNodes() {
if !node.Up { if !node.Up() {
ids = append(ids, node.ID()) ids = append(ids, node.ID())
} }
} }
@ -539,6 +562,10 @@ func (net *Network) getConn(oneID, otherID enode.ID) *Conn {
func (net *Network) InitConn(oneID, otherID enode.ID) (*Conn, error) { func (net *Network) InitConn(oneID, otherID enode.ID) (*Conn, error) {
net.lock.Lock() net.lock.Lock()
defer net.lock.Unlock() defer net.lock.Unlock()
return net.initConn(oneID, otherID)
}
func (net *Network) initConn(oneID, otherID enode.ID) (*Conn, error) {
if oneID == otherID { if oneID == otherID {
return nil, fmt.Errorf("refusing to connect to self %v", oneID) return nil, fmt.Errorf("refusing to connect to self %v", oneID)
} }
@ -602,8 +629,21 @@ type Node struct {
// Config if the config used to created the node // Config if the config used to created the node
Config *adapters.NodeConfig `json:"config"` Config *adapters.NodeConfig `json:"config"`
// Up tracks whether or not the node is running // up tracks whether or not the node is running
Up bool `json:"up"` up bool
upMu sync.RWMutex
}
func (n *Node) Up() bool {
n.upMu.RLock()
defer n.upMu.RUnlock()
return n.up
}
func (n *Node) SetUp(up bool) {
n.upMu.Lock()
defer n.upMu.Unlock()
n.up = up
} }
// ID returns the ID of the node // ID returns the ID of the node
@ -637,10 +677,29 @@ func (n *Node) MarshalJSON() ([]byte, error) {
}{ }{
Info: n.NodeInfo(), Info: n.NodeInfo(),
Config: n.Config, Config: n.Config,
Up: n.Up, Up: n.Up(),
}) })
} }
// UnmarshalJSON implements json.Unmarshaler interface so that we don't lose
// Node.up status. IMPORTANT: The implementation is incomplete; we lose
// p2p.NodeInfo.
func (n *Node) UnmarshalJSON(raw []byte) error {
// TODO: How should we turn back NodeInfo into n.Node?
// Ticket: https://github.com/ethersphere/go-ethereum/issues/1177
node := struct {
Config *adapters.NodeConfig `json:"config,omitempty"`
Up bool `json:"up"`
}{}
if err := json.Unmarshal(raw, &node); err != nil {
return err
}
n.SetUp(node.Up)
n.Config = node.Config
return nil
}
// Conn represents a connection between two nodes in the network // Conn represents a connection between two nodes in the network
type Conn struct { type Conn struct {
// One is the node which initiated the connection // One is the node which initiated the connection
@ -660,10 +719,10 @@ type Conn struct {
// nodesUp returns whether both nodes are currently up // nodesUp returns whether both nodes are currently up
func (c *Conn) nodesUp() error { func (c *Conn) nodesUp() error {
if !c.one.Up { if !c.one.Up() {
return fmt.Errorf("one %v is not up", c.One) return fmt.Errorf("one %v is not up", c.One)
} }
if !c.other.Up { if !c.other.Up() {
return fmt.Errorf("other %v is not up", c.Other) return fmt.Errorf("other %v is not up", c.Other)
} }
return nil return nil
@ -735,7 +794,7 @@ func (net *Network) snapshot(addServices []string, removeServices []string) (*Sn
} }
for i, node := range net.Nodes { for i, node := range net.Nodes {
snap.Nodes[i] = NodeSnapshot{Node: *node} snap.Nodes[i] = NodeSnapshot{Node: *node}
if !node.Up { if !node.Up() {
continue continue
} }
snapshots, err := node.Snapshots() snapshots, err := node.Snapshots()
@ -790,7 +849,7 @@ func (net *Network) Load(snap *Snapshot) error {
if _, err := net.NewNodeWithConfig(n.Node.Config); err != nil { if _, err := net.NewNodeWithConfig(n.Node.Config); err != nil {
return err return err
} }
if !n.Node.Up { if !n.Node.Up() {
continue continue
} }
if err := net.startWithSnapshots(n.Node.Config.ID, n.Snapshots); err != nil { if err := net.startWithSnapshots(n.Node.Config.ID, n.Snapshots); err != nil {
@ -862,7 +921,7 @@ func (net *Network) Load(snap *Snapshot) error {
// Start connecting. // Start connecting.
for _, conn := range snap.Conns { for _, conn := range snap.Conns {
if !net.GetNode(conn.One).Up || !net.GetNode(conn.Other).Up { if !net.GetNode(conn.One).Up() || !net.GetNode(conn.Other).Up() {
//in this case, at least one of the nodes of a connection is not up, //in this case, at least one of the nodes of a connection is not up,
//so it would result in the snapshot `Load` to fail //so it would result in the snapshot `Load` to fail
continue continue
@ -916,7 +975,7 @@ func (net *Network) executeControlEvent(event *Event) {
} }
func (net *Network) executeNodeEvent(e *Event) error { func (net *Network) executeNodeEvent(e *Event) error {
if !e.Node.Up { if !e.Node.Up() {
return net.Stop(e.Node.ID()) return net.Stop(e.Node.ID())
} }

View File

@ -20,6 +20,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"reflect"
"strconv" "strconv"
"strings" "strings"
"testing" "testing"
@ -485,3 +486,137 @@ func benchmarkMinimalServiceTmp(b *testing.B) {
} }
} }
} }
func TestNode_UnmarshalJSON(t *testing.T) {
t.Run(
"test unmarshal of Node up field",
func(t *testing.T) {
runNodeUnmarshalJSON(t, casesNodeUnmarshalJSONUpField())
},
)
t.Run(
"test unmarshal of Node Config field",
func(t *testing.T) {
runNodeUnmarshalJSON(t, casesNodeUnmarshalJSONConfigField())
},
)
}
func runNodeUnmarshalJSON(t *testing.T, tests []nodeUnmarshalTestCase) {
t.Helper()
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
var got Node
if err := got.UnmarshalJSON([]byte(tt.marshaled)); err != nil {
expectErrorMessageToContain(t, err, tt.wantErr)
}
expectNodeEquality(t, got, tt.want)
})
}
}
type nodeUnmarshalTestCase struct {
name string
marshaled string
want Node
wantErr string
}
func expectErrorMessageToContain(t *testing.T, got error, want string) {
t.Helper()
if got == nil && want == "" {
return
}
if got == nil && want != "" {
t.Errorf("error was expected, got: nil, want: %v", want)
return
}
if !strings.Contains(got.Error(), want) {
t.Errorf(
"unexpected error message, got %v, want: %v",
want,
got,
)
}
}
func expectNodeEquality(t *testing.T, got Node, want Node) {
t.Helper()
if !reflect.DeepEqual(got, want) {
t.Errorf("Node.UnmarshalJSON() = %v, want %v", got, want)
}
}
func casesNodeUnmarshalJSONUpField() []nodeUnmarshalTestCase {
return []nodeUnmarshalTestCase{
{
name: "empty json",
marshaled: "{}",
want: Node{
up: false,
},
},
{
name: "a stopped node",
marshaled: "{\"up\": false}",
want: Node{
up: false,
},
},
{
name: "a running node",
marshaled: "{\"up\": true}",
want: Node{
up: true,
},
},
{
name: "invalid JSON value on valid key",
marshaled: "{\"up\": foo}",
wantErr: "invalid character",
},
{
name: "invalid JSON key and value",
marshaled: "{foo: bar}",
wantErr: "invalid character",
},
{
name: "bool value expected but got something else (string)",
marshaled: "{\"up\": \"true\"}",
wantErr: "cannot unmarshal string into Go struct",
},
}
}
func casesNodeUnmarshalJSONConfigField() []nodeUnmarshalTestCase {
// Don't do a big fuss around testing, as adapters.NodeConfig should
// handle it's own serialization. Just do a sanity check.
return []nodeUnmarshalTestCase{
{
name: "Config field is omitted",
marshaled: "{}",
want: Node{
Config: nil,
},
},
{
name: "Config field is nil",
marshaled: "{\"config\": nil}",
want: Node{
Config: nil,
},
},
{
name: "a non default Config field",
marshaled: "{\"config\":{\"name\":\"node_ecdd0\",\"port\":44665}}",
want: Node{
Config: &adapters.NodeConfig{
Name: "node_ecdd0",
Port: 44665,
},
},
},
}
}

View File

@ -44,7 +44,7 @@ func (s *Simulation) NodeIDs() (ids []enode.ID) {
func (s *Simulation) UpNodeIDs() (ids []enode.ID) { func (s *Simulation) UpNodeIDs() (ids []enode.ID) {
nodes := s.Net.GetNodes() nodes := s.Net.GetNodes()
for _, node := range nodes { for _, node := range nodes {
if node.Up { if node.Up() {
ids = append(ids, node.ID()) ids = append(ids, node.ID())
} }
} }
@ -55,7 +55,7 @@ func (s *Simulation) UpNodeIDs() (ids []enode.ID) {
func (s *Simulation) DownNodeIDs() (ids []enode.ID) { func (s *Simulation) DownNodeIDs() (ids []enode.ID) {
nodes := s.Net.GetNodes() nodes := s.Net.GetNodes()
for _, node := range nodes { for _, node := range nodes {
if !node.Up { if !node.Up() {
ids = append(ids, node.ID()) ids = append(ids, node.ID())
} }
} }

View File

@ -54,7 +54,7 @@ func TestUpDownNodeIDs(t *testing.T) {
gotIDs = sim.UpNodeIDs() gotIDs = sim.UpNodeIDs()
for _, id := range gotIDs { for _, id := range gotIDs {
if !sim.Net.GetNode(id).Up { if !sim.Net.GetNode(id).Up() {
t.Errorf("node %s should not be down", id) t.Errorf("node %s should not be down", id)
} }
} }
@ -66,7 +66,7 @@ func TestUpDownNodeIDs(t *testing.T) {
gotIDs = sim.DownNodeIDs() gotIDs = sim.DownNodeIDs()
for _, id := range gotIDs { for _, id := range gotIDs {
if sim.Net.GetNode(id).Up { if sim.Net.GetNode(id).Up() {
t.Errorf("node %s should not be up", id) t.Errorf("node %s should not be up", id)
} }
} }
@ -112,7 +112,7 @@ func TestAddNode(t *testing.T) {
t.Fatal("node not found") t.Fatal("node not found")
} }
if !n.Up { if !n.Up() {
t.Error("node not started") t.Error("node not started")
} }
} }
@ -327,7 +327,7 @@ func TestStartStopNode(t *testing.T) {
if n == nil { if n == nil {
t.Fatal("node not found") t.Fatal("node not found")
} }
if !n.Up { if !n.Up() {
t.Error("node not started") t.Error("node not started")
} }
@ -335,26 +335,17 @@ func TestStartStopNode(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
if n.Up { if n.Up() {
t.Error("node not stopped") t.Error("node not stopped")
} }
// Sleep here to ensure that Network.watchPeerEvents defer function waitForPeerEventPropagation()
// has set the `node.Up = false` before we start the node again.
// p2p/simulations/network.go:215
//
// The same node is stopped and started again, and upon start
// watchPeerEvents is started in a goroutine. If the node is stopped
// and then very quickly started, that goroutine may be scheduled later
// then start and force `node.Up = false` in its defer function.
// This will make this test unreliable.
time.Sleep(time.Second)
err = sim.StartNode(id) err = sim.StartNode(id)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
if !n.Up { if !n.Up() {
t.Error("node not started") t.Error("node not started")
} }
} }
@ -377,7 +368,7 @@ func TestStartStopRandomNode(t *testing.T) {
if n == nil { if n == nil {
t.Fatal("node not found") t.Fatal("node not found")
} }
if n.Up { if n.Up() {
t.Error("node not stopped") t.Error("node not stopped")
} }
@ -386,16 +377,7 @@ func TestStartStopRandomNode(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
// Sleep here to ensure that Network.watchPeerEvents defer function waitForPeerEventPropagation()
// has set the `node.Up = false` before we start the node again.
// p2p/simulations/network.go:215
//
// The same node is stopped and started again, and upon start
// watchPeerEvents is started in a goroutine. If the node is stopped
// and then very quickly started, that goroutine may be scheduled later
// then start and force `node.Up = false` in its defer function.
// This will make this test unreliable.
time.Sleep(time.Second)
idStarted, err := sim.StartRandomNode() idStarted, err := sim.StartRandomNode()
if err != nil { if err != nil {
@ -426,21 +408,12 @@ func TestStartStopRandomNodes(t *testing.T) {
if n == nil { if n == nil {
t.Fatal("node not found") t.Fatal("node not found")
} }
if n.Up { if n.Up() {
t.Error("node not stopped") t.Error("node not stopped")
} }
} }
// Sleep here to ensure that Network.watchPeerEvents defer function waitForPeerEventPropagation()
// has set the `node.Up = false` before we start the node again.
// p2p/simulations/network.go:215
//
// The same node is stopped and started again, and upon start
// watchPeerEvents is started in a goroutine. If the node is stopped
// and then very quickly started, that goroutine may be scheduled later
// then start and force `node.Up = false` in its defer function.
// This will make this test unreliable.
time.Sleep(time.Second)
ids, err = sim.StartRandomNodes(2) ids, err = sim.StartRandomNodes(2)
if err != nil { if err != nil {
@ -452,8 +425,20 @@ func TestStartStopRandomNodes(t *testing.T) {
if n == nil { if n == nil {
t.Fatal("node not found") t.Fatal("node not found")
} }
if !n.Up { if !n.Up() {
t.Error("node not started") t.Error("node not started")
} }
} }
} }
func waitForPeerEventPropagation() {
// Sleep here to ensure that Network.watchPeerEvents defer function
// has set the `node.Up() = false` before we start the node again.
//
// The same node is stopped and started again, and upon start
// watchPeerEvents is started in a goroutine. If the node is stopped
// and then very quickly started, that goroutine may be scheduled later
// then start and force `node.Up() = false` in its defer function.
// This will make this test unreliable.
time.Sleep(1 * time.Second)
}

View File

@ -52,7 +52,7 @@ func (s *Simulation) Services(name string) (services map[enode.ID]node.Service)
nodes := s.Net.GetNodes() nodes := s.Net.GetNodes()
services = make(map[enode.ID]node.Service) services = make(map[enode.ID]node.Service)
for _, node := range nodes { for _, node := range nodes {
if !node.Up { if !node.Up() {
continue continue
} }
simNode, ok := node.Node.(*adapters.SimNode) simNode, ok := node.Node.(*adapters.SimNode)

View File

@ -124,7 +124,7 @@ func TestClose(t *testing.T) {
var upNodeCount int var upNodeCount int
for _, n := range sim.Net.GetNodes() { for _, n := range sim.Net.GetNodes() {
if n.Up { if n.Up() {
upNodeCount++ upNodeCount++
} }
} }
@ -140,7 +140,7 @@ func TestClose(t *testing.T) {
upNodeCount = 0 upNodeCount = 0
for _, n := range sim.Net.GetNodes() { for _, n := range sim.Net.GetNodes() {
if n.Up { if n.Up() {
upNodeCount++ upNodeCount++
} }
} }

View File

@ -178,7 +178,7 @@ func watchSimEvents(net *simulations.Network, ctx context.Context, trigger chan
case ev := <-events: case ev := <-events:
//only catch node up events //only catch node up events
if ev.Type == simulations.EventTypeNode { if ev.Type == simulations.EventTypeNode {
if ev.Node.Up { if ev.Node.Up() {
log.Debug("got node up event", "event", ev, "node", ev.Node.Config.ID) log.Debug("got node up event", "event", ev, "node", ev.Node.Config.ID)
select { select {
case trigger <- ev.Node.Config.ID: case trigger <- ev.Node.Config.ID:

View File

@ -88,18 +88,15 @@ func (s *DBStore) Get(key string, i interface{}) (err error) {
// Put stores an object that implements Binary for a specific key. // Put stores an object that implements Binary for a specific key.
func (s *DBStore) Put(key string, i interface{}) (err error) { func (s *DBStore) Put(key string, i interface{}) (err error) {
var bytes []byte var bytes []byte
if marshaler, ok := i.(encoding.BinaryMarshaler); ok {
marshaler, ok := i.(encoding.BinaryMarshaler)
if !ok {
if bytes, err = json.Marshal(i); err != nil {
return err
}
} else {
if bytes, err = marshaler.MarshalBinary(); err != nil { if bytes, err = marshaler.MarshalBinary(); err != nil {
return err return err
} }
} else {
if bytes, err = json.Marshal(i); err != nil {
return err
}
} }
return s.db.Put([]byte(key), bytes, nil) return s.db.Put([]byte(key), bytes, nil)
} }