tornadocontrib/go-ethereum
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

swarm: network simulation for swarm tests (#769)

Browse Source 
* cmd/swarm: minor cli flag text adjustments

* cmd/swarm, swarm/storage, swarm: fix  mingw on windows test issues

* cmd/swarm: support for smoke tests on the production swarm cluster

* cmd/swarm/swarm-smoke: simplify cluster logic as per suggestion

* changed colour of landing page

* landing page reacts to enter keypress

* swarm/api/http: sticky footer for swarm landing page using flex

* swarm/api/http: sticky footer for error pages and fix for multiple choices

* swarm: propagate ctx to internal apis (#754)

* swarm/simnet: add basic node/service functions

* swarm/netsim: add buckets for global state and kademlia health check

* swarm/netsim: Use sync.Map as bucket and provide cleanup function for...

* swarm, swarm/netsim: adjust SwarmNetworkTest

* swarm/netsim: fix tests

* swarm: added visualization option to sim net redesign

* swarm/netsim: support multiple services per node

* swarm/netsim: remove redundant return statement

* swarm/netsim: add comments

* swarm: shutdown HTTP in Simulation.Close

* swarm: sim HTTP server timeout

* swarm/netsim: add more simulation methods and peer events examples

* swarm/netsim: add WaitKademlia example

* swarm/netsim: fix comments

* swarm/netsim: terminate peer events goroutines on simulation done

* swarm, swarm/netsim: naming updates

* swarm/netsim: return not healthy kademlias on WaitTillHealthy

* swarm: fix WaitTillHealthy call in testSwarmNetwork

* swarm/netsim: allow bucket to have any type for a key

* swarm: Added snapshots to new netsim

* swarm/netsim: add more tests for bucket

* swarm/netsim: move http related things into separate files

* swarm/netsim: add AddNodeWithService option

* swarm/netsim: add more tests and Start* methods

* swarm/netsim: add peer events and kademlia tests

* swarm/netsim: fix some tests flakiness

* swarm/netsim: improve random nodes selection, fix TestStartStop* tests

* swarm/netsim: remove time measurement from TestClose to avoid flakiness

* swarm/netsim: builder pattern for netsim HTTP server (#773)

* swarm/netsim: add connect related tests

* swarm/netsim: add comment for TestPeerEvents

* swarm: rename netsim package to network/simulation
This commit is contained in:
Janoš Guljaš 2018-07-17 07:08:34 +02:00 committed by Anton Evangelatov
parent f5b128a5b3
commit dcaaa3c804
19 changed files with 2861 additions and 200 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
p2p/simulations/adapters/inproc.go
View File

@ -296,6 +296,13 @@ func (sn *SimNode) Stop() error {
return sn.node.Stop() return sn.node.Stop()
} }
// Service returns a running service by name
func (sn *SimNode) Service(name string) node.Service {
sn.lock.RLock()
defer sn.lock.RUnlock()
return sn.running[name]
}
// Services returns a copy of the underlying services // Services returns a copy of the underlying services
func (sn *SimNode) Services() []node.Service { func (sn *SimNode) Services() []node.Service {
sn.lock.RLock() sn.lock.RLock()
@ -307,6 +314,17 @@ func (sn *SimNode) Services() []node.Service {
return services return services
} }
// ServiceMap returns a map by names of the underlying services
func (sn *SimNode) ServiceMap() map[string]node.Service {
sn.lock.RLock()
defer sn.lock.RUnlock()
services := make(map[string]node.Service, len(sn.running))
for name, service := range sn.running {
services[name] = service
}
return services
}
// Server returns the underlying p2p.Server // Server returns the underlying p2p.Server
func (sn *SimNode) Server() *p2p.Server { func (sn *SimNode) Server() *p2p.Server {
return sn.node.Server() return sn.node.Server()

81
swarm/network/simulation/bucket.go Normal file
View File

@ -0,0 +1,81 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"github.com/ethereum/go-ethereum/p2p/discover"
)
// BucketKey is the type that should be used for keys in simulation buckets.
type BucketKey string
// NodeItem returns an item set in ServiceFunc function for a particualar node.
func (s *Simulation) NodeItem(id discover.NodeID, key interface{}) (value interface{}, ok bool) {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.buckets[id]; !ok {
return nil, false
}
return s.buckets[id].Load(key)
}
// SetNodeItem sets a new item associated with the node with provided NodeID.
// Buckets should be used to avoid managing separate simulation global state.
func (s *Simulation) SetNodeItem(id discover.NodeID, key interface{}, value interface{}) {
s.mu.Lock()
defer s.mu.Unlock()
s.buckets[id].Store(key, value)
}
// NodeItems returns a map of items from all nodes that are all set under the
// same BucketKey.
func (s *Simulation) NodesItems(key interface{}) (values map[discover.NodeID]interface{}) {
s.mu.RLock()
defer s.mu.RUnlock()
ids := s.NodeIDs()
values = make(map[discover.NodeID]interface{}, len(ids))
for _, id := range ids {
if _, ok := s.buckets[id]; !ok {
continue
}
if v, ok := s.buckets[id].Load(key); ok {
values[id] = v
}
}
return values
}
// UpNodesItems returns a map of items with the same BucketKey from all nodes that are up.
func (s *Simulation) UpNodesItems(key interface{}) (values map[discover.NodeID]interface{}) {
s.mu.RLock()
defer s.mu.RUnlock()
ids := s.UpNodeIDs()
values = make(map[discover.NodeID]interface{})
for _, id := range ids {
if _, ok := s.buckets[id]; !ok {
continue
}
if v, ok := s.buckets[id].Load(key); ok {
values[id] = v
}
}
return values
}

155
swarm/network/simulation/bucket_test.go Normal file
View File

@ -0,0 +1,155 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"sync"
"testing"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
)
// TestServiceBucket tests all bucket functionalities using subtests.
// It constructs a simulation of two nodes by adding items to their buckets
// in ServiceFunc constructor, then by SetNodeItem. Testing UpNodesItems
// is done by stopping one node and validating availability of its items.
func TestServiceBucket(t *testing.T) {
testKey := "Key"
testValue := "Value"
sim := New(map[string]ServiceFunc{
"noop": func(ctx *adapters.ServiceContext, b *sync.Map) (node.Service, func(), error) {
b.Store(testKey, testValue+ctx.Config.ID.String())
return newNoopService(), nil, nil
},
})
defer sim.Close()
id1, err := sim.AddNode()
if err != nil {
t.Fatal(err)
}
id2, err := sim.AddNode()
if err != nil {
t.Fatal(err)
}
t.Run("ServiceFunc bucket Store", func(t *testing.T) {
v, ok := sim.NodeItem(id1, testKey)
if !ok {
t.Fatal("bucket item not found")
}
s, ok := v.(string)
if !ok {
t.Fatal("bucket item value is not string")
}
if s != testValue+id1.String() {
t.Fatalf("expected %q, got %q", testValue+id1.String(), s)
}
v, ok = sim.NodeItem(id2, testKey)
if !ok {
t.Fatal("bucket item not found")
}
s, ok = v.(string)
if !ok {
t.Fatal("bucket item value is not string")
}
if s != testValue+id2.String() {
t.Fatalf("expected %q, got %q", testValue+id2.String(), s)
}
})
customKey := "anotherKey"
customValue := "anotherValue"
t.Run("SetNodeItem", func(t *testing.T) {
sim.SetNodeItem(id1, customKey, customValue)
v, ok := sim.NodeItem(id1, customKey)
if !ok {
t.Fatal("bucket item not found")
}
s, ok := v.(string)
if !ok {
t.Fatal("bucket item value is not string")
}
if s != customValue {
t.Fatalf("expected %q, got %q", customValue, s)
}
v, ok = sim.NodeItem(id2, customKey)
if ok {
t.Fatal("bucket item should not be found")
}
})
if err := sim.StopNode(id2); err != nil {
t.Fatal(err)
}
t.Run("UpNodesItems", func(t *testing.T) {
items := sim.UpNodesItems(testKey)
v, ok := items[id1]
if !ok {
t.Errorf("node 1 item not found")
}
s, ok := v.(string)
if !ok {
t.Fatal("node 1 item value is not string")
}
if s != testValue+id1.String() {
t.Fatalf("expected %q, got %q", testValue+id1.String(), s)
}
v, ok = items[id2]
if ok {
t.Errorf("node 2 item should not be found")
}
})
t.Run("NodeItems", func(t *testing.T) {
items := sim.NodesItems(testKey)
v, ok := items[id1]
if !ok {
t.Errorf("node 1 item not found")
}
s, ok := v.(string)
if !ok {
t.Fatal("node 1 item value is not string")
}
if s != testValue+id1.String() {
t.Fatalf("expected %q, got %q", testValue+id1.String(), s)
}
v, ok = items[id2]
if !ok {
t.Errorf("node 2 item not found")
}
s, ok = v.(string)
if !ok {
t.Fatal("node 1 item value is not string")
}
if s != testValue+id2.String() {
t.Fatalf("expected %q, got %q", testValue+id2.String(), s)
}
})
}

159
swarm/network/simulation/connect.go Normal file
View File

@ -0,0 +1,159 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"strings"
"github.com/ethereum/go-ethereum/p2p/discover"
)
// ConnectToPivotNode connects the node with provided NodeID
// to the pivot node, already set by Simulation.SetPivotNode method.
// It is useful when constructing a star network topology
// when simulation adds and removes nodes dynamically.
func (s *Simulation) ConnectToPivotNode(id discover.NodeID) (err error) {
pid := s.PivotNodeID()
if pid == nil {
return ErrNoPivotNode
}
return s.connect(*pid, id)
}
// ConnectToLastNode connects the node with provided NodeID
// to the last node that is up, and avoiding connection to self.
// It is useful when constructing a chain network topology
// when simulation adds and removes nodes dynamically.
func (s *Simulation) ConnectToLastNode(id discover.NodeID) (err error) {
ids := s.UpNodeIDs()
l := len(ids)
if l < 2 {
return nil
}
lid := ids[l-1]
if lid == id {
lid = ids[l-2]
}
return s.connect(lid, id)
}
// ConnectToRandomNode connects the node with provieded NodeID
// to a random node that is up.
func (s *Simulation) ConnectToRandomNode(id discover.NodeID) (err error) {
n := s.randomUpNode(id)
if n == nil {
return ErrNodeNotFound
}
return s.connect(n.ID, id)
}
// ConnectNodesFull connects all nodes one to another.
// It provides a complete connectivity in the network
// which should be rarely needed.
func (s *Simulation) ConnectNodesFull(ids []discover.NodeID) (err error) {
if ids == nil {
ids = s.UpNodeIDs()
}
l := len(ids)
for i := 0; i < l; i++ {
for j := i + 1; j < l; j++ {
err = s.connect(ids[i], ids[j])
if err != nil {
return err
}
}
}
return nil
}
// ConnectNodesChain connects all nodes in a chain topology.
// If ids argument is nil, all nodes that are up will be connected.
func (s *Simulation) ConnectNodesChain(ids []discover.NodeID) (err error) {
if ids == nil {
ids = s.UpNodeIDs()
}
l := len(ids)
for i := 0; i < l-1; i++ {
err = s.connect(ids[i], ids[i+1])
if err != nil {
return err
}
}
return nil
}
// ConnectNodesRing connects all nodes in a ring topology.
// If ids argument is nil, all nodes that are up will be connected.
func (s *Simulation) ConnectNodesRing(ids []discover.NodeID) (err error) {
if ids == nil {
ids = s.UpNodeIDs()
}
l := len(ids)
if l < 2 {
return nil
}
for i := 0; i < l-1; i++ {
err = s.connect(ids[i], ids[i+1])
if err != nil {
return err
}
}
return s.connect(ids[l-1], ids[0])
}
// ConnectNodesStar connects all nodes in a star topology
// with the center at provided NodeID.
// If ids argument is nil, all nodes that are up will be connected.
func (s *Simulation) ConnectNodesStar(id discover.NodeID, ids []discover.NodeID) (err error) {
if ids == nil {
ids = s.UpNodeIDs()
}
l := len(ids)
for i := 0; i < l; i++ {
if id == ids[i] {
continue
}
err = s.connect(id, ids[i])
if err != nil {
return err
}
}
return nil
}
// ConnectNodesStar connects all nodes in a star topology
// with the center at already set pivot node.
// If ids argument is nil, all nodes that are up will be connected.
func (s *Simulation) ConnectNodesStarPivot(ids []discover.NodeID) (err error) {
id := s.PivotNodeID()
if id == nil {
return ErrNoPivotNode
}
return s.ConnectNodesStar(*id, ids)
}
// connect connects two nodes but ignores already connected error.
func (s *Simulation) connect(oneID, otherID discover.NodeID) error {
return ignoreAlreadyConnectedErr(s.Net.Connect(oneID, otherID))
}
func ignoreAlreadyConnectedErr(err error) error {
if err == nil || strings.Contains(err.Error(), "already connected") {
return nil
}
return err
}

306
swarm/network/simulation/connect_test.go Normal file
View File

@ -0,0 +1,306 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"testing"
"github.com/ethereum/go-ethereum/p2p/discover"
)
func TestConnectToPivotNode(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
pid, err := sim.AddNode()
if err != nil {
t.Fatal(err)
}
sim.SetPivotNode(pid)
id, err := sim.AddNode()
if err != nil {
t.Fatal(err)
}
if len(sim.Net.Conns) > 0 {
t.Fatal("no connections should exist after just adding nodes")
}
err = sim.ConnectToPivotNode(id)
if err != nil {
t.Fatal(err)
}
if sim.Net.GetConn(id, pid) == nil {
t.Error("node did not connect to pivot node")
}
}
func TestConnectToLastNode(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
n := 10
ids, err := sim.AddNodes(n)
if err != nil {
t.Fatal(err)
}
id, err := sim.AddNode()
if err != nil {
t.Fatal(err)
}
if len(sim.Net.Conns) > 0 {
t.Fatal("no connections should exist after just adding nodes")
}
err = sim.ConnectToLastNode(id)
if err != nil {
t.Fatal(err)
}
for _, i := range ids[:n-2] {
if sim.Net.GetConn(id, i) != nil {
t.Error("node connected to the node that is not the last")
}
}
if sim.Net.GetConn(id, ids[n-1]) == nil {
t.Error("node did not connect to the last node")
}
}
func TestConnectToRandomNode(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
n := 10
ids, err := sim.AddNodes(n)
if err != nil {
t.Fatal(err)
}
if len(sim.Net.Conns) > 0 {
t.Fatal("no connections should exist after just adding nodes")
}
err = sim.ConnectToRandomNode(ids[0])
if err != nil {
t.Fatal(err)
}
var cc int
for i := 0; i < n; i++ {
for j := i + 1; j < n; j++ {
if sim.Net.GetConn(ids[i], ids[j]) != nil {
cc++
}
}
}
if cc != 1 {
t.Errorf("expected one connection, got %v", cc)
}
}
func TestConnectNodesFull(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
ids, err := sim.AddNodes(12)
if err != nil {
t.Fatal(err)
}
if len(sim.Net.Conns) > 0 {
t.Fatal("no connections should exist after just adding nodes")
}
err = sim.ConnectNodesFull(ids)
if err != nil {
t.Fatal(err)
}
testFull(t, sim, ids)
}
func testFull(t *testing.T, sim *Simulation, ids []discover.NodeID) {
n := len(ids)
var cc int
for i := 0; i < n; i++ {
for j := i + 1; j < n; j++ {
if sim.Net.GetConn(ids[i], ids[j]) != nil {
cc++
}
}
}
want := n * (n - 1) / 2
if cc != want {
t.Errorf("expected %v connection, got %v", want, cc)
}
}
func TestConnectNodesChain(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
ids, err := sim.AddNodes(10)
if err != nil {
t.Fatal(err)
}
if len(sim.Net.Conns) > 0 {
t.Fatal("no connections should exist after just adding nodes")
}
err = sim.ConnectNodesChain(ids)
if err != nil {
t.Fatal(err)
}
testChain(t, sim, ids)
}
func testChain(t *testing.T, sim *Simulation, ids []discover.NodeID) {
n := len(ids)
for i := 0; i < n; i++ {
for j := i + 1; j < n; j++ {
c := sim.Net.GetConn(ids[i], ids[j])
if i == j-1 {
if c == nil {
t.Errorf("nodes %v and %v are not connected, but they should be", i, j)
}
} else {
if c != nil {
t.Errorf("nodes %v and %v are connected, but they should not be", i, j)
}
}
}
}
}
func TestConnectNodesRing(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
ids, err := sim.AddNodes(10)
if err != nil {
t.Fatal(err)
}
if len(sim.Net.Conns) > 0 {
t.Fatal("no connections should exist after just adding nodes")
}
err = sim.ConnectNodesRing(ids)
if err != nil {
t.Fatal(err)
}
testRing(t, sim, ids)
}
func testRing(t *testing.T, sim *Simulation, ids []discover.NodeID) {
n := len(ids)
for i := 0; i < n; i++ {
for j := i + 1; j < n; j++ {
c := sim.Net.GetConn(ids[i], ids[j])
if i == j-1 || (i == 0 && j == n-1) {
if c == nil {
t.Errorf("nodes %v and %v are not connected, but they should be", i, j)
}
} else {
if c != nil {
t.Errorf("nodes %v and %v are connected, but they should not be", i, j)
}
}
}
}
}
func TestConnectToNodesStar(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
ids, err := sim.AddNodes(10)
if err != nil {
t.Fatal(err)
}
if len(sim.Net.Conns) > 0 {
t.Fatal("no connections should exist after just adding nodes")
}
centerIndex := 2
err = sim.ConnectNodesStar(ids[centerIndex], ids)
if err != nil {
t.Fatal(err)
}
testStar(t, sim, ids, centerIndex)
}
func testStar(t *testing.T, sim *Simulation, ids []discover.NodeID, centerIndex int) {
n := len(ids)
for i := 0; i < n; i++ {
for j := i + 1; j < n; j++ {
c := sim.Net.GetConn(ids[i], ids[j])
if i == centerIndex || j == centerIndex {
if c == nil {
t.Errorf("nodes %v and %v are not connected, but they should be", i, j)
}
} else {
if c != nil {
t.Errorf("nodes %v and %v are connected, but they should not be", i, j)
}
}
}
}
}
func TestConnectToNodesStarPivot(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
ids, err := sim.AddNodes(10)
if err != nil {
t.Fatal(err)
}
if len(sim.Net.Conns) > 0 {
t.Fatal("no connections should exist after just adding nodes")
}
pivotIndex := 4
sim.SetPivotNode(ids[pivotIndex])
err = sim.ConnectNodesStarPivot(ids)
if err != nil {
t.Fatal(err)
}
testStar(t, sim, ids, pivotIndex)
}

157
swarm/network/simulation/events.go Normal file
View File

@ -0,0 +1,157 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"context"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p"
)
// PeerEvent is the type of the channel returned by Simulation.PeerEvents.
type PeerEvent struct {
// NodeID is the ID of node that the event is caught on.
NodeID discover.NodeID
// Event is the event that is caught.
Event *p2p.PeerEvent
// Error is the error that may have happened during event watching.
Error error
}
// PeerEventsFilter defines a filter on PeerEvents to exclude messages with
// defined properties. Use PeerEventsFilter methods to set required options.
type PeerEventsFilter struct {
t *p2p.PeerEventType
protocol *string
msgCode *uint64
}
// NewPeerEventsFilter returns a new PeerEventsFilter instance.
func NewPeerEventsFilter() *PeerEventsFilter {
return &PeerEventsFilter{}
}
// Type sets the filter to only one peer event type.
func (f *PeerEventsFilter) Type(t p2p.PeerEventType) *PeerEventsFilter {
f.t = &t
return f
}
// Protocol sets the filter to only one message protocol.
func (f *PeerEventsFilter) Protocol(p string) *PeerEventsFilter {
f.protocol = &p
return f
}
// MsgCode sets the filter to only one msg code.
func (f *PeerEventsFilter) MsgCode(c uint64) *PeerEventsFilter {
f.msgCode = &c
return f
}
// PeerEvents returns a channel of events that are captured by admin peerEvents
// subscription nodes with provided NodeIDs. Additional filters can be set to ignore
// events that are not relevant.
func (s *Simulation) PeerEvents(ctx context.Context, ids []discover.NodeID, filters ...*PeerEventsFilter) <-chan PeerEvent {
eventC := make(chan PeerEvent)
for _, id := range ids {
s.shutdownWG.Add(1)
go func(id discover.NodeID) {
defer s.shutdownWG.Done()
client, err := s.Net.GetNode(id).Client()
if err != nil {
eventC <- PeerEvent{NodeID: id, Error: err}
return
}
events := make(chan *p2p.PeerEvent)
sub, err := client.Subscribe(ctx, "admin", events, "peerEvents")
if err != nil {
eventC <- PeerEvent{NodeID: id, Error: err}
return
}
defer sub.Unsubscribe()
for {
select {
case <-ctx.Done():
if err := ctx.Err(); err != nil {
select {
case eventC <- PeerEvent{NodeID: id, Error: err}:
case <-s.Done():
}
}
return
case <-s.Done():
return
case e := <-events:
match := len(filters) == 0 // if there are no filters match all events
for _, f := range filters {
if f.t != nil && *f.t != e.Type {
continue
}
if f.protocol != nil && *f.protocol != e.Protocol {
continue
}
if f.msgCode != nil && e.MsgCode != nil && *f.msgCode != *e.MsgCode {
continue
}
// all filter parameters matched, break the loop
match = true
break
}
if match {
select {
case eventC <- PeerEvent{NodeID: id, Event: e}:
case <-ctx.Done():
if err := ctx.Err(); err != nil {
select {
case eventC <- PeerEvent{NodeID: id, Error: err}:
case <-s.Done():
}
}
return
case <-s.Done():
return
}
}
case err := <-sub.Err():
if err != nil {
select {
case eventC <- PeerEvent{NodeID: id, Error: err}:
case <-ctx.Done():
if err := ctx.Err(); err != nil {
select {
case eventC <- PeerEvent{NodeID: id, Error: err}:
case <-s.Done():
}
}
return
case <-s.Done():
return
}
}
}
}
}(id)
}
return eventC
}

104
swarm/network/simulation/events_test.go Normal file
View File

@ -0,0 +1,104 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"context"
"sync"
"testing"
"time"
)
// TestPeerEvents creates simulation, adds two nodes,
// register for peer events, connects nodes in a chain
// and waits for the number of connection events to
// be received.
func TestPeerEvents(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
_, err := sim.AddNodes(2)
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
events := sim.PeerEvents(ctx, sim.NodeIDs())
// two nodes -> two connection events
expectedEventCount := 2
var wg sync.WaitGroup
wg.Add(expectedEventCount)
go func() {
for e := range events {
if e.Error != nil {
if e.Error == context.Canceled {
return
}
t.Error(e.Error)
continue
}
wg.Done()
}
}()
err = sim.ConnectNodesChain(sim.NodeIDs())
if err != nil {
t.Fatal(err)
}
wg.Wait()
}
func TestPeerEventsTimeout(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
_, err := sim.AddNodes(2)
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
events := sim.PeerEvents(ctx, sim.NodeIDs())
done := make(chan struct{})
go func() {
for e := range events {
if e.Error == context.Canceled {
return
}
if e.Error == context.DeadlineExceeded {
close(done)
return
} else {
t.Fatal(e.Error)
}
}
}()
select {
case <-time.After(time.Second):
t.Error("no context deadline received")
case <-done:
// all good, context deadline detected
}
}

140
swarm/network/simulation/example_test.go Normal file
View File

@ -0,0 +1,140 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation_test
import (
"context"
"fmt"
"sync"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
)
// Every node can have a Kademlia associated using the node bucket under
// BucketKeyKademlia key. This allows to use WaitTillHealthy to block until
// all nodes have the their Kadmlias healthy.
func ExampleSimulation_WaitTillHealthy() {
sim := simulation.New(map[string]simulation.ServiceFunc{
"bzz": func(ctx *adapters.ServiceContext, b *sync.Map) (node.Service, func(), error) {
addr := network.NewAddrFromNodeID(ctx.Config.ID)
hp := network.NewHiveParams()
hp.Discovery = false
config := &network.BzzConfig{
OverlayAddr: addr.Over(),
UnderlayAddr: addr.Under(),
HiveParams: hp,
}
kad := network.NewKademlia(addr.Over(), network.NewKadParams())
// store kademlia in node's bucket under BucketKeyKademlia
// so that it can be found by WaitTillHealthy method.
b.Store(simulation.BucketKeyKademlia, kad)
return network.NewBzz(config, kad, nil, nil, nil), nil, nil
},
})
defer sim.Close()
_, err := sim.AddNodesAndConnectRing(10)
if err != nil {
// handle error properly...
panic(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
ill, err := sim.WaitTillHealthy(ctx, 2)
if err != nil {
// inspect the latest detected not healthy kademlias
for id, kad := range ill {
fmt.Println("Node", id)
fmt.Println(kad.String())
}
// handle error...
}
// continue with the test
}
// Watch all peer events in the simulation network, buy receiving from a channel.
func ExampleSimulation_PeerEvents() {
sim := simulation.New(nil)
defer sim.Close()
events := sim.PeerEvents(context.Background(), sim.NodeIDs())
go func() {
for e := range events {
if e.Error != nil {
log.Error("peer event", "err", e.Error)
continue
}
log.Info("peer event", "node", e.NodeID, "peer", e.Event.Peer, "msgcode", e.Event.MsgCode)
}
}()
}
// Detect when a nodes drop a peer.
func ExampleSimulation_PeerEvents_disconnections() {
sim := simulation.New(nil)
defer sim.Close()
disconnections := sim.PeerEvents(
context.Background(),
sim.NodeIDs(),
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeDrop),
)
go func() {
for d := range disconnections {
if d.Error != nil {
log.Error("peer drop", "err", d.Error)
continue
}
log.Warn("peer drop", "node", d.NodeID, "peer", d.Event.Peer)
}
}()
}
// Watch multiple types of events or messages. In this case, they differ only
// by MsgCode, but filters can be set for different types or protocols, too.
func ExampleSimulation_PeerEvents_multipleFilters() {
sim := simulation.New(nil)
defer sim.Close()
msgs := sim.PeerEvents(
context.Background(),
sim.NodeIDs(),
// Watch when bzz messages 1 and 4 are received.
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("bzz").MsgCode(1),
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("bzz").MsgCode(4),
)
go func() {
for m := range msgs {
if m.Error != nil {
log.Error("bzz message", "err", m.Error)
continue
}
log.Info("bzz message", "node", m.NodeID, "peer", m.Event.Peer)
}
}()
}

63
swarm/network/simulation/http.go Normal file
View File

@ -0,0 +1,63 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"fmt"
"net/http"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/simulations"
)
// Package defaults.
var (
DefaultHTTPSimAddr = ":8888"
)
//`With`(builder) pattern constructor for Simulation to
//start with a HTTP server
func (s *Simulation) WithServer(addr string) *Simulation {
//assign default addr if nothing provided
if addr == "" {
addr = DefaultHTTPSimAddr
}
log.Info(fmt.Sprintf("Initializing simulation server on %s...", addr))
//initialize the HTTP server
s.handler = simulations.NewServer(s.Net)
s.runC = make(chan struct{})
//add swarm specific routes to the HTTP server
s.addSimulationRoutes()
s.httpSrv = &http.Server{
Addr: addr,
Handler: s.handler,
}
go s.httpSrv.ListenAndServe()
return s
}
//register additional HTTP routes
func (s *Simulation) addSimulationRoutes() {
s.handler.POST("/runsim", s.RunSimulation)
}
// StartNetwork starts all nodes in the network
func (s *Simulation) RunSimulation(w http.ResponseWriter, req *http.Request) {
log.Debug("RunSimulation endpoint running")
s.runC <- struct{}{}
w.WriteHeader(http.StatusOK)
}

104
swarm/network/simulation/http_test.go Normal file
View File

@ -0,0 +1,104 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"context"
"fmt"
"net/http"
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
)
func TestSimulationWithHTTPServer(t *testing.T) {
log.Debug("Init simulation")
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
sim := New(
map[string]ServiceFunc{
"noop": func(_ *adapters.ServiceContext, b *sync.Map) (node.Service, func(), error) {
return newNoopService(), nil, nil
},
}).WithServer(DefaultHTTPSimAddr)
defer sim.Close()
log.Debug("Done.")
_, err := sim.AddNode()
if err != nil {
t.Fatal(err)
}
log.Debug("Starting sim round and let it time out...")
//first test that running without sending to the channel will actually
//block the simulation, so let it time out
result := sim.Run(ctx, func(ctx context.Context, sim *Simulation) error {
log.Debug("Just start the sim without any action and wait for the timeout")
//ensure with a Sleep that simulation doesn't terminate before the timeout
time.Sleep(2 * time.Second)
return nil
})
if result.Error != nil {
if result.Error.Error() == "context deadline exceeded" {
log.Debug("Expected timeout error received")
} else {
t.Fatal(result.Error)
}
}
//now run it again and send the expected signal on the waiting channel,
//then close the simulation
log.Debug("Starting sim round and wait for frontend signal...")
//this time the timeout should be long enough so that it doesn't kick in too early
ctx, cancel2 := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel2()
go sendRunSignal(t)
result = sim.Run(ctx, func(ctx context.Context, sim *Simulation) error {
log.Debug("This run waits for the run signal from `frontend`...")
//ensure with a Sleep that simulation doesn't terminate before the signal is received
time.Sleep(2 * time.Second)
return nil
})
if result.Error != nil {
t.Fatal(result.Error)
}
log.Debug("Test terminated successfully")
}
func sendRunSignal(t *testing.T) {
//We need to first wait for the sim HTTP server to start running...
time.Sleep(2 * time.Second)
//then we can send the signal
log.Debug("Sending run signal to simulation: POST /runsim...")
resp, err := http.Post(fmt.Sprintf("http://localhost%s/runsim", DefaultHTTPSimAddr), "application/json", nil)
if err != nil {
t.Fatalf("Request failed: %v", err)
}
defer resp.Body.Close()
log.Debug("Signal sent")
if resp.StatusCode != http.StatusOK {
t.Fatalf("err %s", resp.Status)
}
}

96
swarm/network/simulation/kademlia.go Normal file
View File

@ -0,0 +1,96 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"context"
"encoding/hex"
"time"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/swarm/network"
)
// BucketKeyKademlia is the key to be used for storing the kademlia
// instance for particuar node, usually inside the ServiceFunc function.
var BucketKeyKademlia BucketKey = "kademlia"
// WaitTillHealthy is blocking until the health of all kademlias is true.
// If error is not nil, a map of kademlia that was found not healthy is returned.
func (s *Simulation) WaitTillHealthy(ctx context.Context, kadMinProxSize int) (ill map[discover.NodeID]*network.Kademlia, err error) {
// Prepare PeerPot map for checking Kademlia health
var ppmap map[string]*network.PeerPot
kademlias := s.kademlias()
addrs := make([][]byte, 0, len(kademlias))
for _, k := range kademlias {
addrs = append(addrs, k.BaseAddr())
}
ppmap = network.NewPeerPotMap(kadMinProxSize, addrs)
// Wait for healthy Kademlia on every node before checking files
ticker := time.NewTicker(200 * time.Millisecond)
defer ticker.Stop()
ill = make(map[discover.NodeID]*network.Kademlia)
for {
select {
case <-ctx.Done():
return ill, ctx.Err()
case <-ticker.C:
for k := range ill {
delete(ill, k)
}
log.Debug("kademlia health check", "addr count", len(addrs))
for id, k := range kademlias {
//PeerPot for this node
addr := common.Bytes2Hex(k.BaseAddr())
pp := ppmap[addr]
//call Healthy RPC
h := k.Healthy(pp)
//print info
log.Debug(k.String())
log.Debug("kademlia", "empty bins", pp.EmptyBins, "gotNN", h.GotNN, "knowNN", h.KnowNN, "full", h.Full)
log.Debug("kademlia", "health", h.GotNN && h.KnowNN && h.Full, "addr", hex.EncodeToString(k.BaseAddr()), "node", id)
log.Debug("kademlia", "ill condition", !h.GotNN || !h.Full, "addr", hex.EncodeToString(k.BaseAddr()), "node", id)
if !h.GotNN || !h.Full {
ill[id] = k
}
}
if len(ill) == 0 {
return nil, nil
}
}
}
}
// kademlias returns all Kademlia instances that are set
// in simulation bucket.
func (s *Simulation) kademlias() (ks map[discover.NodeID]*network.Kademlia) {
items := s.UpNodesItems(BucketKeyKademlia)
ks = make(map[discover.NodeID]*network.Kademlia, len(items))
for id, v := range items {
k, ok := v.(*network.Kademlia)
if !ok {
continue
}
ks[id] = k
}
return ks
}

67
swarm/network/simulation/kademlia_test.go Normal file
View File

@ -0,0 +1,67 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"context"
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/network"
)
func TestWaitTillHealthy(t *testing.T) {
sim := New(map[string]ServiceFunc{
"bzz": func(ctx *adapters.ServiceContext, b *sync.Map) (node.Service, func(), error) {
addr := network.NewAddrFromNodeID(ctx.Config.ID)
hp := network.NewHiveParams()
hp.Discovery = false
config := &network.BzzConfig{
OverlayAddr: addr.Over(),
UnderlayAddr: addr.Under(),
HiveParams: hp,
}
kad := network.NewKademlia(addr.Over(), network.NewKadParams())
// store kademlia in node's bucket under BucketKeyKademlia
// so that it can be found by WaitTillHealthy method.
b.Store(BucketKeyKademlia, kad)
return network.NewBzz(config, kad, nil, nil, nil), nil, nil
},
})
defer sim.Close()
_, err := sim.AddNodesAndConnectRing(10)
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()
ill, err := sim.WaitTillHealthy(ctx, 2)
if err != nil {
for id, kad := range ill {
t.Log("Node", id)
t.Log(kad.String())
}
if err != nil {
t.Fatal(err)
}
}
}

357
swarm/network/simulation/node.go Normal file
View File

@ -0,0 +1,357 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"encoding/json"
"errors"
"io/ioutil"
"math/rand"
"os"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
)
// NodeIDs returns NodeIDs for all nodes in the network.
func (s *Simulation) NodeIDs() (ids []discover.NodeID) {
nodes := s.Net.GetNodes()
ids = make([]discover.NodeID, len(nodes))
for i, node := range nodes {
ids[i] = node.ID()
}
return ids
}
// UpNodeIDs returns NodeIDs for nodes that are up in the network.
func (s *Simulation) UpNodeIDs() (ids []discover.NodeID) {
nodes := s.Net.GetNodes()
for _, node := range nodes {
if node.Up {
ids = append(ids, node.ID())
}
}
return ids
}
// DownNodeIDs returns NodeIDs for nodes that are stopped in the network.
func (s *Simulation) DownNodeIDs() (ids []discover.NodeID) {
nodes := s.Net.GetNodes()
for _, node := range nodes {
if !node.Up {
ids = append(ids, node.ID())
}
}
return ids
}
// AddNodeOption defines the option that can be passed
// to Simulation.AddNode method.
type AddNodeOption func(*adapters.NodeConfig)
// AddNodeWithMsgEvents sets the EnableMsgEvents option
// to NodeConfig.
func AddNodeWithMsgEvents(enable bool) AddNodeOption {
return func(o *adapters.NodeConfig) {
o.EnableMsgEvents = enable
}
}
// AddNodeWithService specifies a service that should be
// started on a node. This option can be repeated as variadic
// argument toe AddNode and other add node related methods.
// If AddNodeWithService is not specified, all services will be started.
func AddNodeWithService(serviceName string) AddNodeOption {
return func(o *adapters.NodeConfig) {
o.Services = append(o.Services, serviceName)
}
}
// AddNode creates a new node with random configuration,
// applies provided options to the config and adds the node to network.
// By default all services will be started on a node. If one or more
// AddNodeWithService option are provided, only specified services will be started.
func (s *Simulation) AddNode(opts ...AddNodeOption) (id discover.NodeID, err error) {
conf := adapters.RandomNodeConfig()
for _, o := range opts {
o(conf)
}
if len(conf.Services) == 0 {
conf.Services = s.serviceNames
}
node, err := s.Net.NewNodeWithConfig(conf)
if err != nil {
return id, err
}
return node.ID(), s.Net.Start(node.ID())
}
// AddNodes creates new nodes with random configurations,
// applies provided options to the config and adds nodes to network.
func (s *Simulation) AddNodes(count int, opts ...AddNodeOption) (ids []discover.NodeID, err error) {
ids = make([]discover.NodeID, 0, count)
for i := 0; i < count; i++ {
id, err := s.AddNode(opts...)
if err != nil {
return nil, err
}
ids = append(ids, id)
}
return ids, nil
}
// AddNodesAndConnectFull is a helpper method that combines
// AddNodes and ConnectNodesFull. Only new nodes will be connected.
func (s *Simulation) AddNodesAndConnectFull(count int, opts ...AddNodeOption) (ids []discover.NodeID, err error) {
if count < 2 {
return nil, errors.New("count of nodes must be at least 2")
}
ids, err = s.AddNodes(count, opts...)
if err != nil {
return nil, err
}
err = s.ConnectNodesFull(ids)
if err != nil {
return nil, err
}
return ids, nil
}
// AddNodesAndConnectChain is a helpper method that combines
// AddNodes and ConnectNodesChain. The chain will be continued from the last
// added node, if there is one in simulation using ConnectToLastNode method.
func (s *Simulation) AddNodesAndConnectChain(count int, opts ...AddNodeOption) (ids []discover.NodeID, err error) {
if count < 2 {
return nil, errors.New("count of nodes must be at least 2")
}
id, err := s.AddNode(opts...)
if err != nil {
return nil, err
}
err = s.ConnectToLastNode(id)
if err != nil {
return nil, err
}
ids, err = s.AddNodes(count-1, opts...)
if err != nil {
return nil, err
}
ids = append([]discover.NodeID{id}, ids...)
err = s.ConnectNodesChain(ids)
if err != nil {
return nil, err
}
return ids, nil
}
// AddNodesAndConnectRing is a helpper method that combines
// AddNodes and ConnectNodesRing.
func (s *Simulation) AddNodesAndConnectRing(count int, opts ...AddNodeOption) (ids []discover.NodeID, err error) {
if count < 2 {
return nil, errors.New("count of nodes must be at least 2")
}
ids, err = s.AddNodes(count, opts...)
if err != nil {
return nil, err
}
err = s.ConnectNodesRing(ids)
if err != nil {
return nil, err
}
return ids, nil
}
// AddNodesAndConnectStar is a helpper method that combines
// AddNodes and ConnectNodesStar.
func (s *Simulation) AddNodesAndConnectStar(count int, opts ...AddNodeOption) (ids []discover.NodeID, err error) {
if count < 2 {
return nil, errors.New("count of nodes must be at least 2")
}
ids, err = s.AddNodes(count, opts...)
if err != nil {
return nil, err
}
err = s.ConnectNodesStar(ids[0], ids[1:])
if err != nil {
return nil, err
}
return ids, nil
}
//Upload a snapshot
//This method tries to open the json file provided, applies the config to all nodes
//and then loads the snapshot into the Simulation network
func (s *Simulation) UploadSnapshot(snapshotFile string, opts ...AddNodeOption) error {
f, err := os.Open(snapshotFile)
if err != nil {
return err
}
defer f.Close()
jsonbyte, err := ioutil.ReadAll(f)
if err != nil {
return err
}
var snap simulations.Snapshot
err = json.Unmarshal(jsonbyte, &snap)
if err != nil {
return err
}
//the snapshot probably has the property EnableMsgEvents not set
//just in case, set it to true!
//(we need this to wait for messages before uploading)
for _, n := range snap.Nodes {
n.Node.Config.EnableMsgEvents = true
n.Node.Config.Services = s.serviceNames
for _, o := range opts {
o(n.Node.Config)
}
}
log.Info("Waiting for p2p connections to be established...")
//now we can load the snapshot
err = s.Net.Load(&snap)
if err != nil {
return err
}
log.Info("Snapshot loaded")
return nil
}
// SetPivotNode sets the NodeID of the network's pivot node.
// Pivot node is just a specific node that should be treated
// differently then other nodes in test. SetPivotNode and
// PivotNodeID are just a convenient functions to set and
// retrieve it.
func (s *Simulation) SetPivotNode(id discover.NodeID) {
s.mu.Lock()
defer s.mu.Unlock()
s.pivotNodeID = &id
}
// PivotNodeID returns NodeID of the pivot node set by
// Simulation.SetPivotNode method.
func (s *Simulation) PivotNodeID() (id *discover.NodeID) {
s.mu.Lock()
defer s.mu.Unlock()
return s.pivotNodeID
}
// StartNode starts a node by NodeID.
func (s *Simulation) StartNode(id discover.NodeID) (err error) {
return s.Net.Start(id)
}
// StartRandomNode starts a random node.
func (s *Simulation) StartRandomNode() (id discover.NodeID, err error) {
n := s.randomDownNode()
if n == nil {
return id, ErrNodeNotFound
}
return n.ID, s.Net.Start(n.ID)
}
// StartRandomNodes starts random nodes.
func (s *Simulation) StartRandomNodes(count int) (ids []discover.NodeID, err error) {
ids = make([]discover.NodeID, 0, count)
downIDs := s.DownNodeIDs()
for i := 0; i < count; i++ {
n := s.randomNode(downIDs, ids...)
if n == nil {
return nil, ErrNodeNotFound
}
err = s.Net.Start(n.ID)
if err != nil {
return nil, err
}
ids = append(ids, n.ID)
}
return ids, nil
}
// StopNode stops a node by NodeID.
func (s *Simulation) StopNode(id discover.NodeID) (err error) {
return s.Net.Stop(id)
}
// StopRandomNode stops a random node.
func (s *Simulation) StopRandomNode() (id discover.NodeID, err error) {
n := s.randomUpNode()
if n == nil {
return id, ErrNodeNotFound
}
return n.ID, s.Net.Stop(n.ID)
}
// StopRandomNodes stops random nodes.
func (s *Simulation) StopRandomNodes(count int) (ids []discover.NodeID, err error) {
ids = make([]discover.NodeID, 0, count)
upIDs := s.UpNodeIDs()
for i := 0; i < count; i++ {
n := s.randomNode(upIDs, ids...)
if n == nil {
return nil, ErrNodeNotFound
}
err = s.Net.Stop(n.ID)
if err != nil {
return nil, err
}
ids = append(ids, n.ID)
}
return ids, nil
}
// seed the random generator for Simulation.randomNode.
func init() {
rand.Seed(time.Now().UnixNano())
}
// randomUpNode returns a random SimNode that is up.
// Arguments are NodeIDs for nodes that should not be returned.
func (s *Simulation) randomUpNode(exclude ...discover.NodeID) *adapters.SimNode {
return s.randomNode(s.UpNodeIDs(), exclude...)
}
// randomUpNode returns a random SimNode that is not up.
func (s *Simulation) randomDownNode(exclude ...discover.NodeID) *adapters.SimNode {
return s.randomNode(s.DownNodeIDs(), exclude...)
}
// randomUpNode returns a random SimNode from the slice of NodeIDs.
func (s *Simulation) randomNode(ids []discover.NodeID, exclude ...discover.NodeID) *adapters.SimNode {
for _, e := range exclude {
var i int
for _, id := range ids {
if id == e {
ids = append(ids[:i], ids[i+1:]...)
} else {
i++
}
}
}
l := len(ids)
if l == 0 {
return nil
}
n := s.Net.GetNode(ids[rand.Intn(l)])
node, _ := n.Node.(*adapters.SimNode)
return node
}

462
swarm/network/simulation/node_test.go Normal file
View File

@ -0,0 +1,462 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"context"
"fmt"
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/network"
)
func TestUpDownNodeIDs(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
ids, err := sim.AddNodes(10)
if err != nil {
t.Fatal(err)
}
gotIDs := sim.NodeIDs()
if !equalNodeIDs(ids, gotIDs) {
t.Error("returned nodes are not equal to added ones")
}
stoppedIDs, err := sim.StopRandomNodes(3)
if err != nil {
t.Fatal(err)
}
gotIDs = sim.UpNodeIDs()
for _, id := range gotIDs {
if !sim.Net.GetNode(id).Up {
t.Errorf("node %s should not be down", id)
}
}
if !equalNodeIDs(ids, append(gotIDs, stoppedIDs...)) {
t.Error("returned nodes are not equal to added ones")
}
gotIDs = sim.DownNodeIDs()
for _, id := range gotIDs {
if sim.Net.GetNode(id).Up {
t.Errorf("node %s should not be up", id)
}
}
if !equalNodeIDs(stoppedIDs, gotIDs) {
t.Error("returned nodes are not equal to the stopped ones")
}
}
func equalNodeIDs(one, other []discover.NodeID) bool {
if len(one) != len(other) {
return false
}
var count int
for _, a := range one {
var found bool
for _, b := range other {
if a == b {
found = true
break
}
}
if found {
count++
} else {
return false
}
}
return count == len(one)
}
func TestAddNode(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
id, err := sim.AddNode()
if err != nil {
t.Fatal(err)
}
n := sim.Net.GetNode(id)
if n == nil {
t.Fatal("node not found")
}
if !n.Up {
t.Error("node not started")
}
}
func TestAddNodeWithMsgEvents(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
id, err := sim.AddNode(AddNodeWithMsgEvents(true))
if err != nil {
t.Fatal(err)
}
if !sim.Net.GetNode(id).Config.EnableMsgEvents {
t.Error("EnableMsgEvents is false")
}
id, err = sim.AddNode(AddNodeWithMsgEvents(false))
if err != nil {
t.Fatal(err)
}
if sim.Net.GetNode(id).Config.EnableMsgEvents {
t.Error("EnableMsgEvents is true")
}
}
func TestAddNodeWithService(t *testing.T) {
sim := New(map[string]ServiceFunc{
"noop1": noopServiceFunc,
"noop2": noopServiceFunc,
})
defer sim.Close()
id, err := sim.AddNode(AddNodeWithService("noop1"))
if err != nil {
t.Fatal(err)
}
n := sim.Net.GetNode(id).Node.(*adapters.SimNode)
if n.Service("noop1") == nil {
t.Error("service noop1 not found on node")
}
if n.Service("noop2") != nil {
t.Error("service noop2 should not be found on node")
}
}
func TestAddNodes(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
nodesCount := 12
ids, err := sim.AddNodes(nodesCount)
if err != nil {
t.Fatal(err)
}
count := len(ids)
if count != nodesCount {
t.Errorf("expected %v nodes, got %v", nodesCount, count)
}
count = len(sim.Net.GetNodes())
if count != nodesCount {
t.Errorf("expected %v nodes, got %v", nodesCount, count)
}
}
func TestAddNodesAndConnectFull(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
n := 12
ids, err := sim.AddNodesAndConnectFull(n)
if err != nil {
t.Fatal(err)
}
testFull(t, sim, ids)
}
func TestAddNodesAndConnectChain(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
_, err := sim.AddNodesAndConnectChain(12)
if err != nil {
t.Fatal(err)
}
// add another set of nodes to test
// if two chains are connected
_, err = sim.AddNodesAndConnectChain(7)
if err != nil {
t.Fatal(err)
}
testChain(t, sim, sim.UpNodeIDs())
}
func TestAddNodesAndConnectRing(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
ids, err := sim.AddNodesAndConnectRing(12)
if err != nil {
t.Fatal(err)
}
testRing(t, sim, ids)
}
func TestAddNodesAndConnectStar(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
ids, err := sim.AddNodesAndConnectStar(12)
if err != nil {
t.Fatal(err)
}
testStar(t, sim, ids, 0)
}
//To test that uploading a snapshot works
func TestUploadSnapshot(t *testing.T) {
log.Debug("Creating simulation")
s := New(map[string]ServiceFunc{
"bzz": func(ctx *adapters.ServiceContext, b *sync.Map) (node.Service, func(), error) {
addr := network.NewAddrFromNodeID(ctx.Config.ID)
hp := network.NewHiveParams()
hp.Discovery = false
config := &network.BzzConfig{
OverlayAddr: addr.Over(),
UnderlayAddr: addr.Under(),
HiveParams: hp,
}
kad := network.NewKademlia(addr.Over(), network.NewKadParams())
return network.NewBzz(config, kad, nil, nil, nil), nil, nil
},
})
defer s.Close()
nodeCount := 16
log.Debug("Uploading snapshot")
err := s.UploadSnapshot(fmt.Sprintf("../stream/testing/snapshot_%d.json", nodeCount))
if err != nil {
t.Fatalf("Error uploading snapshot to simulation network: %v", err)
}
ctx := context.Background()
log.Debug("Starting simulation...")
s.Run(ctx, func(ctx context.Context, sim *Simulation) error {
log.Debug("Checking")
nodes := sim.UpNodeIDs()
if len(nodes) != nodeCount {
t.Fatal("Simulation network node number doesn't match snapshot node number")
}
return nil
})
log.Debug("Done.")
}
func TestPivotNode(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
id, err := sim.AddNode()
if err != nil {
t.Fatal(err)
}
id2, err := sim.AddNode()
if err != nil {
t.Fatal(err)
}
if sim.PivotNodeID() != nil {
t.Error("expected no pivot node")
}
sim.SetPivotNode(id)
pid := sim.PivotNodeID()
if pid == nil {
t.Error("pivot node not set")
} else if *pid != id {
t.Errorf("expected pivot node %s, got %s", id, *pid)
}
sim.SetPivotNode(id2)
pid = sim.PivotNodeID()
if pid == nil {
t.Error("pivot node not set")
} else if *pid != id2 {
t.Errorf("expected pivot node %s, got %s", id2, *pid)
}
}
func TestStartStopNode(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
id, err := sim.AddNode()
if err != nil {
t.Fatal(err)
}
n := sim.Net.GetNode(id)
if n == nil {
t.Fatal("node not found")
}
if !n.Up {
t.Error("node not started")
}
err = sim.StopNode(id)
if err != nil {
t.Fatal(err)
}
if n.Up {
t.Error("node not stopped")
}
// Sleep here to ensure that Network.watchPeerEvents defer function
// 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)
if err != nil {
t.Fatal(err)
}
if !n.Up {
t.Error("node not started")
}
}
func TestStartStopRandomNode(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
_, err := sim.AddNodes(3)
if err != nil {
t.Fatal(err)
}
id, err := sim.StopRandomNode()
if err != nil {
t.Fatal(err)
}
n := sim.Net.GetNode(id)
if n == nil {
t.Fatal("node not found")
}
if n.Up {
t.Error("node not stopped")
}
id2, err := sim.StopRandomNode()
if err != nil {
t.Fatal(err)
}
// Sleep here to ensure that Network.watchPeerEvents defer function
// 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()
if err != nil {
t.Fatal(err)
}
if idStarted != id && idStarted != id2 {
t.Error("unexpected started node ID")
}
}
func TestStartStopRandomNodes(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
_, err := sim.AddNodes(10)
if err != nil {
t.Fatal(err)
}
ids, err := sim.StopRandomNodes(3)
if err != nil {
t.Fatal(err)
}
for _, id := range ids {
n := sim.Net.GetNode(id)
if n == nil {
t.Fatal("node not found")
}
if n.Up {
t.Error("node not stopped")
}
}
// Sleep here to ensure that Network.watchPeerEvents defer function
// 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)
if err != nil {
t.Fatal(err)
}
for _, id := range ids {
n := sim.Net.GetNode(id)
if n == nil {
t.Fatal("node not found")
}
if !n.Up {
t.Error("node not started")
}
}
}

65
swarm/network/simulation/service.go Normal file
View File

@ -0,0 +1,65 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
)
// Service returns a single Service by name on a particular node
// with provided id.
func (s *Simulation) Service(name string, id discover.NodeID) node.Service {
simNode, ok := s.Net.GetNode(id).Node.(*adapters.SimNode)
if !ok {
return nil
}
services := simNode.ServiceMap()
if len(services) == 0 {
return nil
}
return services[name]
}
// RandomService returns a single Service by name on a
// randomly chosen node that is up.
func (s *Simulation) RandomService(name string) node.Service {
n := s.randomUpNode()
if n == nil {
return nil
}
return n.Service(name)
}
// Services returns all services with a provided name
// from nodes that are up.
func (s *Simulation) Services(name string) (services map[discover.NodeID]node.Service) {
nodes := s.Net.GetNodes()
services = make(map[discover.NodeID]node.Service)
for _, node := range nodes {
if !node.Up {
continue
}
simNode, ok := node.Node.(*adapters.SimNode)
if !ok {
continue
}
services[node.ID()] = simNode.Service(name)
}
return services
}

46
swarm/network/simulation/service_test.go Normal file
View File

@ -0,0 +1,46 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"testing"
)
func TestService(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
id, err := sim.AddNode()
if err != nil {
t.Fatal(err)
}
_, ok := sim.Service("noop", id).(*noopService)
if !ok {
t.Fatalf("service is not of %T type", &noopService{})
}
_, ok = sim.RandomService("noop").(*noopService)
if !ok {
t.Fatalf("service is not of %T type", &noopService{})
}
_, ok = sim.Services("noop")[id].(*noopService)
if !ok {
t.Fatalf("service is not of %T type", &noopService{})
}
}

201
swarm/network/simulation/simulation.go Normal file
View File

@ -0,0 +1,201 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"context"
"errors"
"net/http"
"sync"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
)
// Common errors that are returned by functions in this package.
var (
ErrNodeNotFound = errors.New("node not found")
ErrNoPivotNode = errors.New("no pivot node set")
)
// Simulation provides methods on network, nodes and services
// to manage them.
type Simulation struct {
// Net is exposed as a way to access lower level functionalities
// of p2p/simulations.Network.
Net *simulations.Network
serviceNames []string
cleanupFuncs []func()
buckets map[discover.NodeID]*sync.Map
pivotNodeID *discover.NodeID
shutdownWG sync.WaitGroup
done chan struct{}
mu sync.RWMutex
httpSrv *http.Server //attach a HTTP server via SimulationOptions
handler *simulations.Server //HTTP handler for the server
runC chan struct{} //channel where frontend signals it is ready
}
// ServiceFunc is used in New to declare new service constructor.
// The first argument provides ServiceContext from the adapters package
// giving for example the access to NodeID. Second argument is the sync.Map
// where all "global" state related to the service should be kept.
// All cleanups needed for constructed service and any other constructed
// objects should ne provided in a single returned cleanup function.
type ServiceFunc func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error)
// New creates a new Simulation instance with new
// simulations.Network initialized with provided services.
func New(services map[string]ServiceFunc) (s *Simulation) {
s = &Simulation{
buckets: make(map[discover.NodeID]*sync.Map),
done: make(chan struct{}),
}
adapterServices := make(map[string]adapters.ServiceFunc, len(services))
for name, serviceFunc := range services {
s.serviceNames = append(s.serviceNames, name)
adapterServices[name] = func(ctx *adapters.ServiceContext) (node.Service, error) {
b := new(sync.Map)
service, cleanup, err := serviceFunc(ctx, b)
if err != nil {
return nil, err
}
s.mu.Lock()
defer s.mu.Unlock()
if cleanup != nil {
s.cleanupFuncs = append(s.cleanupFuncs, cleanup)
}
s.buckets[ctx.Config.ID] = b
return service, nil
}
}
s.Net = simulations.NewNetwork(
adapters.NewSimAdapter(adapterServices),
&simulations.NetworkConfig{ID: "0"},
)
return s
}
// RunFunc is the function that will be called
// on Simulation.Run method call.
type RunFunc func(context.Context, *Simulation) error
// Result is the returned value of Simulation.Run method.
type Result struct {
Duration time.Duration
Error error
}
// Run calls the RunFunc function while taking care of
// cancelation provided through the Context.
func (s *Simulation) Run(ctx context.Context, f RunFunc) (r Result) {
//if the option is set to run a HTTP server with the simulation,
//init the server and start it
start := time.Now()
if s.httpSrv != nil {
log.Info("Waiting for frontend to be ready...(send POST /runsim to HTTP server)")
//wait for the frontend to connect
select {
case <-s.runC:
case <-ctx.Done():
return Result{
Duration: time.Since(start),
Error: ctx.Err(),
}
}
log.Info("Received signal from frontend - starting simulation run.")
}
errc := make(chan error)
quit := make(chan struct{})
defer close(quit)
go func() {
select {
case errc <- f(ctx, s):
case <-quit:
}
}()
var err error
select {
case <-ctx.Done():
err = ctx.Err()
case err = <-errc:
}
return Result{
Duration: time.Since(start),
Error: err,
}
}
// Maximal number of parallel calls to cleanup functions on
// Simulation.Close.
var maxParallelCleanups = 10
// Close calls all cleanup functions that are returned by
// ServiceFunc, waits for all of them to finish and other
// functions that explicitly block shutdownWG
// (like Simulation.PeerEvents) and shuts down the network
// at the end. It is used to clean all resources from the
// simulation.
func (s *Simulation) Close() {
close(s.done)
sem := make(chan struct{}, maxParallelCleanups)
s.mu.RLock()
cleanupFuncs := make([]func(), len(s.cleanupFuncs))
for i, f := range s.cleanupFuncs {
if f != nil {
cleanupFuncs[i] = f
}
}
s.mu.RUnlock()
for _, cleanup := range cleanupFuncs {
s.shutdownWG.Add(1)
sem <- struct{}{}
go func(cleanup func()) {
defer s.shutdownWG.Done()
defer func() { <-sem }()
cleanup()
}(cleanup)
}
if s.httpSrv != nil {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
err := s.httpSrv.Shutdown(ctx)
if err != nil {
log.Error("Error shutting down HTTP server!", "err", err)
}
close(s.runC)
}
s.shutdownWG.Wait()
s.Net.Shutdown()
}
// Done returns a channel that is closed when the simulation
// is closed by Close method. It is useful for signaling termination
// of all possible goroutines that are created within the test.
func (s *Simulation) Done() <-chan struct{} {
return s.done
}

207
swarm/network/simulation/simulation_test.go Normal file
View File

@ -0,0 +1,207 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package simulation
import (
"context"
"errors"
"flag"
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rpc"
colorable "github.com/mattn/go-colorable"
)
var (
loglevel = flag.Int("loglevel", 2, "verbosity of logs")
)
func init() {
flag.Parse()
log.PrintOrigins(true)
log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true))))
}
// TestRun tests if Run method calls RunFunc and if it handles context properly.
func TestRun(t *testing.T) {
sim := New(noopServiceFuncMap)
defer sim.Close()
t.Run("call", func(t *testing.T) {
expect := "something"
var got string
r := sim.Run(context.Background(), func(ctx context.Context, sim *Simulation) error {
got = expect
return nil
})
if r.Error != nil {
t.Errorf("unexpected error: %v", r.Error)
}
if got != expect {
t.Errorf("expected %q, got %q", expect, got)
}
})
t.Run("cancelation", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
r := sim.Run(ctx, func(ctx context.Context, sim *Simulation) error {
time.Sleep(100 * time.Millisecond)
return nil
})
if r.Error != context.DeadlineExceeded {
t.Errorf("unexpected error: %v", r.Error)
}
})
t.Run("context value and duration", func(t *testing.T) {
ctx := context.WithValue(context.Background(), "hey", "there")
sleep := 50 * time.Millisecond
r := sim.Run(ctx, func(ctx context.Context, sim *Simulation) error {
if ctx.Value("hey") != "there" {
return errors.New("expected context value not passed")
}
time.Sleep(sleep)
return nil
})
if r.Error != nil {
t.Errorf("unexpected error: %v", r.Error)
}
if r.Duration < sleep {
t.Errorf("reported run duration less then expected: %s", r.Duration)
}
})
}
// TestClose tests are Close method triggers all close functions and are all nodes not up anymore.
func TestClose(t *testing.T) {
var mu sync.Mutex
var cleanupCount int
sleep := 50 * time.Millisecond
sim := New(map[string]ServiceFunc{
"noop": func(ctx *adapters.ServiceContext, b *sync.Map) (node.Service, func(), error) {
return newNoopService(), func() {
time.Sleep(sleep)
mu.Lock()
defer mu.Unlock()
cleanupCount++
}, nil
},
})
nodeCount := 30
_, err := sim.AddNodes(nodeCount)
if err != nil {
t.Fatal(err)
}
var upNodeCount int
for _, n := range sim.Net.GetNodes() {
if n.Up {
upNodeCount++
}
}
if upNodeCount != nodeCount {
t.Errorf("all nodes should be up, insted only %v are up", upNodeCount)
}
sim.Close()
if cleanupCount != nodeCount {
t.Errorf("number of cleanups expected %v, got %v", nodeCount, cleanupCount)
}
upNodeCount = 0
for _, n := range sim.Net.GetNodes() {
if n.Up {
upNodeCount++
}
}
if upNodeCount != 0 {
t.Errorf("all nodes should be down, insted %v are up", upNodeCount)
}
}
// TestDone checks if Close method triggers the closing of done channel.
func TestDone(t *testing.T) {
sim := New(noopServiceFuncMap)
sleep := 50 * time.Millisecond
timeout := 2 * time.Second
start := time.Now()
go func() {
time.Sleep(sleep)
sim.Close()
}()
select {
case <-time.After(timeout):
t.Error("done channel closing timmed out")
case <-sim.Done():
if d := time.Since(start); d < sleep {
t.Errorf("done channel closed sooner then expected: %s", d)
}
}
}
// a helper map for usual services that do not do anyting
var noopServiceFuncMap = map[string]ServiceFunc{
"noop": noopServiceFunc,
}
// a helper function for most basic noop service
func noopServiceFunc(ctx *adapters.ServiceContext, b *sync.Map) (node.Service, func(), error) {
return newNoopService(), nil, nil
}
// noopService is the service that does not do anything
// but implements node.Service interface.
type noopService struct{}
func newNoopService() node.Service {
return &noopService{}
}
func (t *noopService) Protocols() []p2p.Protocol {
return []p2p.Protocol{}
}
func (t *noopService) APIs() []rpc.API {
return []rpc.API{}
}
func (t *noopService) Start(server *p2p.Server) error {
return nil
}
func (t *noopService) Stop() error {
return nil
}

273
swarm/network_test.go
View File

@ -28,15 +28,14 @@ import (
"testing" "testing"
"time" "time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/api" "github.com/ethereum/go-ethereum/swarm/api"
"github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
"github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage"
colorable "github.com/mattn/go-colorable" colorable "github.com/mattn/go-colorable"
) )
@ -261,16 +260,46 @@ type testSwarmNetworkOptions struct {
// - May wait for Kademlia on every node to be healthy. // - May wait for Kademlia on every node to be healthy.
// - Checking if a file is retrievable from all nodes. // - Checking if a file is retrievable from all nodes.
func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwarmNetworkStep) { func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwarmNetworkStep) {
dir, err := ioutil.TempDir("", "swarm-network-test")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(dir)
if o == nil { if o == nil {
o = new(testSwarmNetworkOptions) o = new(testSwarmNetworkOptions)
} }
sim := simulation.New(map[string]simulation.ServiceFunc{
"swarm": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
config := api.NewConfig()
dir, err := ioutil.TempDir("", "swarm-network-test-node")
if err != nil {
return nil, nil, err
}
cleanup = func() {
err := os.RemoveAll(dir)
if err != nil {
log.Error("cleaning up swarm temp dir", "err", err)
}
}
config.Path = dir
privkey, err := crypto.GenerateKey()
if err != nil {
return nil, cleanup, err
}
config.Init(privkey)
config.DeliverySkipCheck = o.SkipCheck
swarm, err := NewSwarm(config, nil)
if err != nil {
return nil, cleanup, err
}
bucket.Store(simulation.BucketKeyKademlia, swarm.bzz.Hive.Overlay.(*network.Kademlia))
log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()))
return swarm, cleanup, nil
},
})
defer sim.Close()
ctx := context.Background() ctx := context.Background()
if o.Timeout > 0 { if o.Timeout > 0 {
var cancel context.CancelFunc var cancel context.CancelFunc
@ -278,61 +307,20 @@ func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwa
defer cancel() defer cancel()
} }
swarms := make(map[discover.NodeID]*Swarm)
files := make([]file, 0) files := make([]file, 0)
services := map[string]adapters.ServiceFunc{
"swarm": func(ctx *adapters.ServiceContext) (node.Service, error) {
config := api.NewConfig()
dir, err := ioutil.TempDir(dir, "node")
if err != nil {
return nil, err
}
config.Path = dir
privkey, err := crypto.GenerateKey()
if err != nil {
return nil, err
}
config.Init(privkey)
config.DeliverySkipCheck = o.SkipCheck
s, err := NewSwarm(config, nil)
if err != nil {
return nil, err
}
log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", s.bzz.BaseAddr()))
swarms[ctx.Config.ID] = s
return s, nil
},
}
a := adapters.NewSimAdapter(services)
net := simulations.NewNetwork(a, &simulations.NetworkConfig{
ID: "0",
DefaultService: "swarm",
})
defer net.Shutdown()
trigger := make(chan discover.NodeID)
sim := simulations.NewSimulation(net)
for i, step := range steps { for i, step := range steps {
log.Debug("test sync step", "n", i+1, "nodes", step.nodeCount) log.Debug("test sync step", "n", i+1, "nodes", step.nodeCount)
change := step.nodeCount - len(allNodeIDs(net)) change := step.nodeCount - len(sim.UpNodeIDs())
if change > 0 { if change > 0 {
_, err := addNodes(change, net) _, err := sim.AddNodesAndConnectChain(change)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
} else if change < 0 { } else if change < 0 {
err := removeNodes(-change, net) _, err := sim.StopRandomNodes(-change)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -341,91 +329,43 @@ func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwa
continue continue
} }
nodeIDs := allNodeIDs(net)
shuffle(len(nodeIDs), func(i, j int) {
nodeIDs[i], nodeIDs[j] = nodeIDs[j], nodeIDs[i]
})
for _, id := range nodeIDs {
key, data, err := uploadFile(swarms[id])
if err != nil {
t.Fatal(err)
}
log.Trace("file uploaded", "node", id, "key", key.String())
files = append(files, file{
addr: key,
data: data,
nodeID: id,
})
}
// Prepare PeerPot map for checking Kademlia health
var ppmap map[string]*network.PeerPot
nIDs := allNodeIDs(net)
addrs := make([][]byte, len(nIDs))
if *waitKademlia {
for i, id := range nIDs {
addrs[i] = swarms[id].bzz.BaseAddr()
}
ppmap = network.NewPeerPotMap(2, addrs)
}
var checkStatusM sync.Map var checkStatusM sync.Map
var nodeStatusM sync.Map var nodeStatusM sync.Map
var totalFoundCount uint64 var totalFoundCount uint64
result := sim.Run(ctx, &simulations.Step{ result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
Action: func(ctx context.Context) error { nodeIDs := sim.UpNodeIDs()
if *waitKademlia { shuffle(len(nodeIDs), func(i, j int) {
// Wait for healthy Kademlia on every node before checking files nodeIDs[i], nodeIDs[j] = nodeIDs[j], nodeIDs[i]
ticker := time.NewTicker(200 * time.Millisecond) })
defer ticker.Stop() for _, id := range nodeIDs {
key, data, err := uploadFile(sim.Service("swarm", id).(*Swarm))
for range ticker.C { if err != nil {
healthy := true return err
log.Debug("kademlia health check", "node count", len(nIDs), "addr count", len(addrs))
for i, id := range nIDs {
swarm := swarms[id]
//PeerPot for this node
addr := common.Bytes2Hex(swarm.bzz.BaseAddr())
pp := ppmap[addr]
//call Healthy RPC
h := swarm.bzz.Healthy(pp)
//print info
log.Debug(swarm.bzz.String())
log.Debug("kademlia", "empty bins", pp.EmptyBins, "gotNN", h.GotNN, "knowNN", h.KnowNN, "full", h.Full)
log.Debug("kademlia", "health", h.GotNN && h.KnowNN && h.Full, "addr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()), "id", id, "i", i)
log.Debug("kademlia", "ill condition", !h.GotNN || !h.Full, "addr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()), "id", id, "i", i)
if !h.GotNN || !h.Full {
healthy = false
break
}
}
if healthy {
break
}
}
} }
log.Trace("file uploaded", "node", id, "key", key.String())
files = append(files, file{
addr: key,
data: data,
nodeID: id,
})
}
go func() { if *waitKademlia {
// File retrieval check is repeated until all uploaded files are retrieved from all nodes if _, err := sim.WaitTillHealthy(ctx, 2); err != nil {
// or until the timeout is reached. return err
for { }
if retrieve(net, files, swarms, trigger, &checkStatusM, &nodeStatusM, &totalFoundCount) == 0 { }
return
} // File retrieval check is repeated until all uploaded files are retrieved from all nodes
} // or until the timeout is reached.
}() for {
return nil if retrieve(sim, files, &checkStatusM, &nodeStatusM, &totalFoundCount) == 0 {
}, return nil
Trigger: trigger, }
Expect: &simulations.Expectation{ }
Nodes: allNodeIDs(net),
Check: func(ctx context.Context, id discover.NodeID) (bool, error) {
// The check is done by a goroutine in the action function.
return true, nil
},
},
}) })
if result.Error != nil { if result.Error != nil {
t.Fatal(result.Error) t.Fatal(result.Error)
} }
@ -433,70 +373,6 @@ func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwa
} }
} }
// allNodeIDs is returning NodeID for every node that is Up.
func allNodeIDs(net *simulations.Network) (nodes []discover.NodeID) {
for _, n := range net.GetNodes() {
if n.Up {
nodes = append(nodes, n.ID())
}
}
return
}
// addNodes adds a number of nodes to the network.
func addNodes(count int, net *simulations.Network) (ids []discover.NodeID, err error) {
for i := 0; i < count; i++ {
nodeIDs := allNodeIDs(net)
l := len(nodeIDs)
nodeconf := adapters.RandomNodeConfig()
node, err := net.NewNodeWithConfig(nodeconf)
if err != nil {
return nil, fmt.Errorf("create node: %v", err)
}
err = net.Start(node.ID())
if err != nil {
return nil, fmt.Errorf("start node: %v", err)
}
log.Debug("created node", "id", node.ID())
// connect nodes in a chain
if l > 0 {
var otherNodeID discover.NodeID
for i := l - 1; i >= 0; i-- {
n := net.GetNode(nodeIDs[i])
if n.Up {
otherNodeID = n.ID()
break
}
}
log.Debug("connect nodes", "one", node.ID(), "other", otherNodeID)
if err := net.Connect(node.ID(), otherNodeID); err != nil {
return nil, err
}
}
ids = append(ids, node.ID())
}
return ids, nil
}
// removeNodes stops a random nodes in the network.
func removeNodes(count int, net *simulations.Network) error {
for i := 0; i < count; i++ {
// allNodeIDs are returning only the Up nodes.
nodeIDs := allNodeIDs(net)
if len(nodeIDs) == 0 {
break
}
node := net.GetNode(nodeIDs[rand.Intn(len(nodeIDs))])
if err := node.Stop(); err != nil {
return err
}
log.Debug("removed node", "id", node.ID())
}
return nil
}
// uploadFile, uploads a short file to the swarm instance // uploadFile, uploads a short file to the swarm instance
// using the api.Put method. // using the api.Put method.
func uploadFile(swarm *Swarm) (storage.Address, string, error) { func uploadFile(swarm *Swarm) (storage.Address, string, error) {
@ -522,10 +398,8 @@ func uploadFile(swarm *Swarm) (storage.Address, string, error) {
// retrieve is the function that is used for checking the availability of // retrieve is the function that is used for checking the availability of
// uploaded files in testSwarmNetwork test helper function. // uploaded files in testSwarmNetwork test helper function.
func retrieve( func retrieve(
net *simulations.Network, sim *simulation.Simulation,
files []file, files []file,
swarms map[discover.NodeID]*Swarm,
trigger chan discover.NodeID,
checkStatusM *sync.Map, checkStatusM *sync.Map,
nodeStatusM *sync.Map, nodeStatusM *sync.Map,
totalFoundCount *uint64, totalFoundCount *uint64,
@ -537,7 +411,7 @@ func retrieve(
var totalWg sync.WaitGroup var totalWg sync.WaitGroup
errc := make(chan error) errc := make(chan error)
nodeIDs := allNodeIDs(net) nodeIDs := sim.UpNodeIDs()
totalCheckCount := len(nodeIDs) * len(files) totalCheckCount := len(nodeIDs) * len(files)
@ -553,8 +427,8 @@ func retrieve(
var wg sync.WaitGroup var wg sync.WaitGroup
swarm := sim.Service("swarm", id).(*Swarm)
for _, f := range files { for _, f := range files {
swarm := swarms[id]
checkKey := check{ checkKey := check{
key: f.addr.String(), key: f.addr.String(),
@ -601,7 +475,6 @@ func retrieve(
if foundCount == checkCount { if foundCount == checkCount {
log.Info("all files are found for node", "id", id.String(), "duration", time.Since(start)) log.Info("all files are found for node", "id", id.String(), "duration", time.Since(start))
nodeStatusM.Store(id, 0) nodeStatusM.Store(id, 0)
trigger <- id
return return
} }
log.Debug("files missing for node", "id", id.String(), "check", checkCount, "found", foundCount) log.Debug("files missing for node", "id", id.String(), "check", checkCount, "found", foundCount)