diff options
| -rw-r--r-- | p2p/simulations/adapters/inproc.go | 18 | ||||
| -rw-r--r-- | swarm/network/simulation/bucket.go | 81 | ||||
| -rw-r--r-- | swarm/network/simulation/bucket_test.go | 155 | ||||
| -rw-r--r-- | swarm/network/simulation/connect.go | 159 | ||||
| -rw-r--r-- | swarm/network/simulation/connect_test.go | 306 | ||||
| -rw-r--r-- | swarm/network/simulation/events.go | 157 | ||||
| -rw-r--r-- | swarm/network/simulation/events_test.go | 104 | ||||
| -rw-r--r-- | swarm/network/simulation/example_test.go | 140 | ||||
| -rw-r--r-- | swarm/network/simulation/http.go | 63 | ||||
| -rw-r--r-- | swarm/network/simulation/http_test.go | 104 | ||||
| -rw-r--r-- | swarm/network/simulation/kademlia.go | 96 | ||||
| -rw-r--r-- | swarm/network/simulation/kademlia_test.go | 67 | ||||
| -rw-r--r-- | swarm/network/simulation/node.go | 357 | ||||
| -rw-r--r-- | swarm/network/simulation/node_test.go | 462 | ||||
| -rw-r--r-- | swarm/network/simulation/service.go | 65 | ||||
| -rw-r--r-- | swarm/network/simulation/service_test.go | 46 | ||||
| -rw-r--r-- | swarm/network/simulation/simulation.go | 201 | ||||
| -rw-r--r-- | swarm/network/simulation/simulation_test.go | 207 | ||||
| -rw-r--r-- | swarm/network_test.go | 249 |
19 files changed, 2849 insertions, 188 deletions
diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index b68d08f39..c1b032a0d 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -296,6 +296,13 @@ func (sn *SimNode) Stop() error { 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 func (sn *SimNode) Services() []node.Service { sn.lock.RLock() @@ -307,6 +314,17 @@ func (sn *SimNode) Services() []node.Service { 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 func (sn *SimNode) Server() *p2p.Server { return sn.node.Server() 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) |