aboutsummaryrefslogtreecommitdiffstats
path: root/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'swarm')
-rw-r--r--swarm/network/simulation/bucket.go81
-rw-r--r--swarm/network/simulation/bucket_test.go155
-rw-r--r--swarm/network/simulation/connect.go159
-rw-r--r--swarm/network/simulation/connect_test.go306
-rw-r--r--swarm/network/simulation/events.go157
-rw-r--r--swarm/network/simulation/events_test.go104
-rw-r--r--swarm/network/simulation/example_test.go140
-rw-r--r--swarm/network/simulation/http.go63
-rw-r--r--swarm/network/simulation/http_test.go104
-rw-r--r--swarm/network/simulation/kademlia.go96
-rw-r--r--swarm/network/simulation/kademlia_test.go67
-rw-r--r--swarm/network/simulation/node.go357
-rw-r--r--swarm/network/simulation/node_test.go462
-rw-r--r--swarm/network/simulation/service.go65
-rw-r--r--swarm/network/simulation/service_test.go46
-rw-r--r--swarm/network/simulation/simulation.go201
-rw-r--r--swarm/network/simulation/simulation_test.go207
-rw-r--r--swarm/network_test.go249
18 files changed, 2831 insertions, 188 deletions
diff --git a/swarm/network/simulation/bucket.go b/swarm/network/simulation/bucket.go
new file mode 100644
index 000000000..b37afaaa4
--- /dev/null
+++ b/swarm/network/simulation/bucket.go
@@ -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
+}
diff --git a/swarm/network/simulation/bucket_test.go b/swarm/network/simulation/bucket_test.go
new file mode 100644
index 000000000..461d99825
--- /dev/null
+++ b/swarm/network/simulation/bucket_test.go
@@ -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)
+ }
+ })
+}
diff --git a/swarm/network/simulation/connect.go b/swarm/network/simulation/connect.go
new file mode 100644
index 000000000..3fe82052b
--- /dev/null
+++ b/swarm/network/simulation/connect.go
@@ -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
+}
diff --git a/swarm/network/simulation/connect_test.go b/swarm/network/simulation/connect_test.go
new file mode 100644
index 000000000..10d73e4a1
--- /dev/null
+++ b/swarm/network/simulation/connect_test.go
@@ -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)
+}
diff --git a/swarm/network/simulation/events.go b/swarm/network/simulation/events.go
new file mode 100644
index 000000000..f9cfadb73
--- /dev/null
+++ b/swarm/network/simulation/events.go
@@ -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
+}
diff --git a/swarm/network/simulation/events_test.go b/swarm/network/simulation/events_test.go
new file mode 100644
index 000000000..0c185d977
--- /dev/null
+++ b/swarm/network/simulation/events_test.go
@@ -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
+ }
+}
diff --git a/swarm/network/simulation/example_test.go b/swarm/network/simulation/example_test.go
new file mode 100644
index 000000000..2a8116921
--- /dev/null
+++ b/swarm/network/simulation/example_test.go
@@ -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)
+ }
+ }()
+}
diff --git a/swarm/network/simulation/http.go b/swarm/network/simulation/http.go
new file mode 100644
index 000000000..40f13f32d
--- /dev/null
+++ b/swarm/network/simulation/http.go
@@ -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)
+}
diff --git a/swarm/network/simulation/http_test.go b/swarm/network/simulation/http_test.go
new file mode 100644
index 000000000..4d8bf9946
--- /dev/null
+++ b/swarm/network/simulation/http_test.go
@@ -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)
+ }
+}
diff --git a/swarm/network/simulation/kademlia.go b/swarm/network/simulation/kademlia.go
new file mode 100644
index 000000000..3e45cb0ce
--- /dev/null
+++ b/swarm/network/simulation/kademlia.go
@@ -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
+}
diff --git a/swarm/network/simulation/kademlia_test.go b/swarm/network/simulation/kademlia_test.go
new file mode 100644
index 000000000..d11fe7e41
--- /dev/null
+++ b/swarm/network/simulation/kademlia_test.go
@@ -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)
+ }
+ }
+}
diff --git a/swarm/network/simulation/node.go b/swarm/network/simulation/node.go
new file mode 100644
index 000000000..bc433cfd8
--- /dev/null
+++ b/swarm/network/simulation/node.go
@@ -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
+}
diff --git a/swarm/network/simulation/node_test.go b/swarm/network/simulation/node_test.go
new file mode 100644
index 000000000..94f0b4fac
--- /dev/null
+++ b/swarm/network/simulation/node_test.go
@@ -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")
+ }
+ }
+}
diff --git a/swarm/network/simulation/service.go b/swarm/network/simulation/service.go
new file mode 100644
index 000000000..d1cbf1f8b
--- /dev/null
+++ b/swarm/network/simulation/service.go
@@ -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
+}
diff --git a/swarm/network/simulation/service_test.go b/swarm/network/simulation/service_test.go
new file mode 100644
index 000000000..23b0d86f2
--- /dev/null
+++ b/swarm/network/simulation/service_test.go
@@ -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{})
+ }
+}
diff --git a/swarm/network/simulation/simulation.go b/swarm/network/simulation/simulation.go
new file mode 100644
index 000000000..2241dfca2
--- /dev/null
+++ b/swarm/network/simulation/simulation.go
@@ -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
+}
diff --git a/swarm/network/simulation/simulation_test.go b/swarm/network/simulation/simulation_test.go
new file mode 100644
index 000000000..803e0499a
--- /dev/null
+++ b/swarm/network/simulation/simulation_test.go
@@ -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
+}
diff --git a/swarm/network_test.go b/swarm/network_test.go
index 606a83be2..d2a030933 100644
--- a/swarm/network_test.go
+++ b/swarm/network_test.go
@@ -28,15 +28,14 @@ import (
"testing"
"time"
- "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"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"
"github.com/ethereum/go-ethereum/swarm/api"
"github.com/ethereum/go-ethereum/swarm/network"
+ "github.com/ethereum/go-ethereum/swarm/network/simulation"
"github.com/ethereum/go-ethereum/swarm/storage"
colorable "github.com/mattn/go-colorable"
)
@@ -261,78 +260,67 @@ type testSwarmNetworkOptions struct {
// - May wait for Kademlia on every node to be healthy.
// - Checking if a file is retrievable from all nodes.
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 {
o = new(testSwarmNetworkOptions)
}
- ctx := context.Background()
- if o.Timeout > 0 {
- var cancel context.CancelFunc
- ctx, cancel = context.WithTimeout(ctx, o.Timeout)
- defer cancel()
- }
-
- swarms := make(map[discover.NodeID]*Swarm)
- files := make([]file, 0)
-
- services := map[string]adapters.ServiceFunc{
- "swarm": func(ctx *adapters.ServiceContext) (node.Service, error) {
+ 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(dir, "node")
+ dir, err := ioutil.TempDir("", "swarm-network-test-node")
if err != nil {
- return nil, err
+ 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, err
+ return nil, cleanup, err
}
config.Init(privkey)
config.DeliverySkipCheck = o.SkipCheck
- s, err := NewSwarm(config, nil)
+ swarm, err := NewSwarm(config, nil)
if err != nil {
- return nil, err
+ return nil, cleanup, err
}
- log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", s.bzz.BaseAddr()))
- swarms[ctx.Config.ID] = s
- return s, nil
+ 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
},
- }
-
- a := adapters.NewSimAdapter(services)
- net := simulations.NewNetwork(a, &simulations.NetworkConfig{
- ID: "0",
- DefaultService: "swarm",
})
- defer net.Shutdown()
+ defer sim.Close()
- trigger := make(chan discover.NodeID)
+ ctx := context.Background()
+ if o.Timeout > 0 {
+ var cancel context.CancelFunc
+ ctx, cancel = context.WithTimeout(ctx, o.Timeout)
+ defer cancel()
+ }
- sim := simulations.NewSimulation(net)
+ files := make([]file, 0)
for i, step := range steps {
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 {
- _, err := addNodes(change, net)
+ _, err := sim.AddNodesAndConnectChain(change)
if err != nil {
t.Fatal(err)
}
} else if change < 0 {
- err := removeNodes(-change, net)
+ _, err := sim.StopRandomNodes(-change)
if err != nil {
t.Fatal(err)
}
@@ -341,160 +329,48 @@ func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwa
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 nodeStatusM sync.Map
var totalFoundCount uint64
- result := sim.Run(ctx, &simulations.Step{
- Action: func(ctx context.Context) error {
- if *waitKademlia {
- // Wait for healthy Kademlia on every node before checking files
- ticker := time.NewTicker(200 * time.Millisecond)
- defer ticker.Stop()
-
- for range ticker.C {
- healthy := true
- 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
- }
- }
+ result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
+ nodeIDs := sim.UpNodeIDs()
+ shuffle(len(nodeIDs), func(i, j int) {
+ nodeIDs[i], nodeIDs[j] = nodeIDs[j], nodeIDs[i]
+ })
+ for _, id := range nodeIDs {
+ key, data, err := uploadFile(sim.Service("swarm", id).(*Swarm))
+ if err != nil {
+ return err
}
+ log.Trace("file uploaded", "node", id, "key", key.String())
+ files = append(files, file{
+ addr: key,
+ data: data,
+ nodeID: id,
+ })
+ }
- go func() {
- // File retrieval check is repeated until all uploaded files are retrieved from all nodes
- // or until the timeout is reached.
- for {
- if retrieve(net, files, swarms, trigger, &checkStatusM, &nodeStatusM, &totalFoundCount) == 0 {
- return
- }
- }
- }()
- 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 {
- t.Fatal(result.Error)
- }
- log.Debug("done: test sync step", "n", i+1, "nodes", step.nodeCount)
- }
-}
-
-// 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
+ if *waitKademlia {
+ if _, err := sim.WaitTillHealthy(ctx, 2); err != nil {
+ return err
}
}
- log.Debug("connect nodes", "one", node.ID(), "other", otherNodeID)
- if err := net.Connect(node.ID(), otherNodeID); err != nil {
- return nil, err
+
+ // File retrieval check is repeated until all uploaded files are retrieved from all nodes
+ // or until the timeout is reached.
+ for {
+ if retrieve(sim, files, &checkStatusM, &nodeStatusM, &totalFoundCount) == 0 {
+ return nil
+ }
}
- }
- 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
+ if result.Error != nil {
+ t.Fatal(result.Error)
}
- log.Debug("removed node", "id", node.ID())
+ log.Debug("done: test sync step", "n", i+1, "nodes", step.nodeCount)
}
- return nil
}
// uploadFile, uploads a short file to the swarm instance
@@ -522,10 +398,8 @@ func uploadFile(swarm *Swarm) (storage.Address, string, error) {
// retrieve is the function that is used for checking the availability of
// uploaded files in testSwarmNetwork test helper function.
func retrieve(
- net *simulations.Network,
+ sim *simulation.Simulation,
files []file,
- swarms map[discover.NodeID]*Swarm,
- trigger chan discover.NodeID,
checkStatusM *sync.Map,
nodeStatusM *sync.Map,
totalFoundCount *uint64,
@@ -537,7 +411,7 @@ func retrieve(
var totalWg sync.WaitGroup
errc := make(chan error)
- nodeIDs := allNodeIDs(net)
+ nodeIDs := sim.UpNodeIDs()
totalCheckCount := len(nodeIDs) * len(files)
@@ -553,8 +427,8 @@ func retrieve(
var wg sync.WaitGroup
+ swarm := sim.Service("swarm", id).(*Swarm)
for _, f := range files {
- swarm := swarms[id]
checkKey := check{
key: f.addr.String(),
@@ -601,7 +475,6 @@ func retrieve(
if foundCount == checkCount {
log.Info("all files are found for node", "id", id.String(), "duration", time.Since(start))
nodeStatusM.Store(id, 0)
- trigger <- id
return
}
log.Debug("files missing for node", "id", id.String(), "check", checkCount, "found", foundCount)