From 247e40a36a5c9d9bbf690a51cbab74a5aed87793 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Fri, 28 Apr 2017 19:11:11 +0100 Subject: [PATCH] p2p/simulations: Get peer events via RPC Signed-off-by: Lewis Marshall --- p2p/adapters/docker.go | 10 +- p2p/adapters/exec.go | 110 +++++++++++++-- p2p/adapters/inproc.go | 45 +++--- p2p/adapters/types.go | 20 --- p2p/peer.go | 20 +++ p2p/protocols/protocol_test.go | 5 +- p2p/server.go | 13 ++ p2p/simulations/mocker.go | 3 +- p2p/simulations/network.go | 133 ++++++++++++++++-- p2p/testing/protocoltester.go | 33 +++-- swarm/network/hive_test.go | 1 - swarm/network/pss_test.go | 92 ++++++------ .../simulations/discovery/discovery_test.go | 18 ++- swarm/network/simulations/overlay.go | 12 +- 14 files changed, 378 insertions(+), 137 deletions(-) diff --git a/p2p/adapters/docker.go b/p2p/adapters/docker.go index d74dd13b87..cb9c0dbea4 100644 --- a/p2p/adapters/docker.go +++ b/p2p/adapters/docker.go @@ -1,6 +1,7 @@ package adapters import ( + "crypto/ecdsa" "fmt" "io" "io/ioutil" @@ -22,7 +23,7 @@ type DockerNode struct { // NewDockerNode creates a new DockerNode, building the docker image if // necessary -func NewDockerNode(id *NodeId, service string) (*DockerNode, error) { +func NewDockerNode(id *NodeId, key *ecdsa.PrivateKey, service string) (*DockerNode, error) { if runtime.GOOS != "linux" { return nil, fmt.Errorf("NewDockerNode can only be used on Linux as it uses the current binary (which must be a Linux binary)") } @@ -51,6 +52,7 @@ func NewDockerNode(id *NodeId, service string) (*DockerNode, error) { ID: id, Service: service, Config: &conf, + key: key, }, } node.newCmd = node.dockerCommand @@ -60,13 +62,13 @@ func NewDockerNode(id *NodeId, service string) (*DockerNode, error) { // dockerCommand returns a command which exec's the binary in a docker // container. // -// It uses a shell so that we can pass the _P2P_NODE_CONFIG environment -// variable to the container using the --env flag. +// It uses a shell so that we can pass the _P2P_NODE_CONFIG and _P2P_NODE_KEY +// environment variables to the container using the --env flag. func (n *DockerNode) dockerCommand() *exec.Cmd { return exec.Command( "sh", "-c", fmt.Sprintf( - `exec docker run --interactive --env _P2P_NODE_CONFIG="${_P2P_NODE_CONFIG}" %s p2p-node %s %s`, + `exec docker run --interactive --env _P2P_NODE_CONFIG="${_P2P_NODE_CONFIG}" --env _P2P_NODE_KEY="${_P2P_NODE_KEY}" %s p2p-node %s %s`, dockerImage, n.Service, n.ID.String(), ), ) diff --git a/p2p/adapters/exec.go b/p2p/adapters/exec.go index 0348418095..4eecd6b4fc 100644 --- a/p2p/adapters/exec.go +++ b/p2p/adapters/exec.go @@ -2,6 +2,8 @@ package adapters import ( "context" + "crypto/ecdsa" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -16,6 +18,7 @@ import ( "time" "github.com/docker/docker/pkg/reexec" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" @@ -55,11 +58,12 @@ type ExecNode struct { client *rpc.Client newCmd func() *exec.Cmd + key *ecdsa.PrivateKey } // NewExecNode creates a new ExecNode which will run the given service using a // sub-directory of the given baseDir -func NewExecNode(id *NodeId, service, baseDir string) (*ExecNode, error) { +func NewExecNode(id *NodeId, key *ecdsa.PrivateKey, service, baseDir string) (*ExecNode, error) { if _, exists := serviceFuncs[service]; !exists { return nil, fmt.Errorf("unknown node service %q", service) } @@ -82,6 +86,7 @@ func NewExecNode(id *NodeId, service, baseDir string) (*ExecNode, error) { Service: service, Dir: dir, Config: &conf, + key: key, } node.newCmd = node.execCommand return node, nil @@ -99,9 +104,10 @@ func (n *ExecNode) Client() (*rpc.Client, error) { return n.client, nil } -// Start exec's the node passing the ID and service as command line arguments -// and the node config encoded as JSON in the _P2P_NODE_CONFIG environment -// variable +// Start exec's the node passing the ID and service as command line arguments, +// the node config encoded as JSON in the _P2P_NODE_CONFIG environment +// variable and the node's private key hex-endoded in the _P2P_NODE_KEY +// environment variable func (n *ExecNode) Start() (err error) { if n.Cmd != nil { return errors.New("already started") @@ -119,6 +125,9 @@ func (n *ExecNode) Start() (err error) { return fmt.Errorf("error generating node config: %s", err) } + // encode the private key + key := hex.EncodeToString(crypto.FromECDSA(n.key)) + // create a net.Pipe for RPC communication over stdin / stdout pipe1, pipe2 := net.Pipe() @@ -127,7 +136,10 @@ func (n *ExecNode) Start() (err error) { cmd.Stdin = pipe1 cmd.Stdout = pipe1 cmd.Stderr = os.Stderr - cmd.Env = append(os.Environ(), fmt.Sprintf("_P2P_NODE_CONFIG=%s", conf)) + cmd.Env = append(os.Environ(), + fmt.Sprintf("_P2P_NODE_CONFIG=%s", conf), + fmt.Sprintf("_P2P_NODE_KEY=%s", key), + ) if err := cmd.Start(); err != nil { return fmt.Errorf("error starting node: %s", err) } @@ -215,6 +227,17 @@ func execP2PNode() { log.Crit("error decoding _P2P_NODE_CONFIG", "err", err) } + // decode the private key + keyEnv := os.Getenv("_P2P_NODE_KEY") + if keyEnv == "" { + log.Crit("missing _P2P_NODE_KEY") + } + key, err := hex.DecodeString(keyEnv) + if err != nil { + log.Crit("error decoding _P2P_NODE_KEY", "err", err) + } + conf.P2P.PrivateKey = crypto.ToECDSA(key) + // initialize the service serviceFunc, exists := serviceFuncs[serviceName] if !exists { @@ -268,10 +291,22 @@ func startP2PNode(conf *node.Config, service node.Service) (*node.Node, error) { if err != nil { return nil, err } - constructor := func(ctx *node.ServiceContext) (node.Service, error) { - return service, nil + + constructor := func(s node.Service) node.ServiceConstructor { + return func(ctx *node.ServiceContext) (node.Service, error) { + return s, nil + } } - if err := stack.Register(constructor); err != nil { + + // register the peer events API + // + // TODO: move this to node.PrivateAdminAPI once the following is merged: + // https://github.com/ethereum/go-ethereum/pull/13885 + if err := stack.Register(constructor(&PeerAPI{stack.Server})); err != nil { + return nil, err + } + + if err := stack.Register(constructor(service)); err != nil { return nil, err } if err := stack.Start(); err != nil { @@ -280,6 +315,65 @@ func startP2PNode(conf *node.Config, service node.Service) (*node.Node, error) { return stack, nil } +// PeerAPI is used to expose peer events under the "eth" RPC namespace. +// +// TODO: move this to node.PrivateAdminAPI and expose under the "admin" +// namespace once the following is merged: +// https://github.com/ethereum/go-ethereum/pull/13885 +type PeerAPI struct { + server func() p2p.Server +} + +func (p *PeerAPI) Protocols() []p2p.Protocol { + return nil +} + +func (p *PeerAPI) APIs() []rpc.API { + return []rpc.API{{ + Namespace: "eth", + Version: "1.0", + Service: p, + }} +} + +func (p *PeerAPI) Start(p2p.Server) error { + return nil +} + +func (p *PeerAPI) Stop() error { + return nil +} + +// PeerEvents creates an RPC sunscription which receives peer events from the +// underlying p2p.Server +func (p *PeerAPI) PeerEvents(ctx context.Context) (*rpc.Subscription, error) { + notifier, supported := rpc.NotifierFromContext(ctx) + if !supported { + return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported + } + + rpcSub := notifier.CreateSubscription() + + go func() { + events := make(chan *p2p.PeerEvent) + sub := p.server().SubscribePeers(events) + defer sub.Unsubscribe() + + for { + select { + case event := <-events: + notifier.Notify(rpcSub.ID, event) + case <-rpcSub.Err(): + return + case <-notifier.Closed(): + return + } + } + }() + + return rpcSub, nil +} + // stdioConn wraps os.Stdin / os.Stdout with a nop Close method so we can // use them to handle RPC messages type stdioConn struct { diff --git a/p2p/adapters/inproc.go b/p2p/adapters/inproc.go index 7c96cf21ca..740e69bfc5 100644 --- a/p2p/adapters/inproc.go +++ b/p2p/adapters/inproc.go @@ -21,6 +21,7 @@ import ( "fmt" "sync" + "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" @@ -53,13 +54,14 @@ type Network interface { // SimNode is the network adapter that type SimNode struct { - lock sync.RWMutex - Id *NodeId - network Network - service node.Service - peerMap map[discover.NodeID]int - peers []*Peer - client *rpc.Client + lock sync.RWMutex + Id *NodeId + network Network + service node.Service + peerMap map[discover.NodeID]int + peers []*Peer + peerFeed event.Feed + client *rpc.Client } func NewSimNode(id *NodeId, svc node.Service, n Network) *SimNode { @@ -115,13 +117,20 @@ func (self *SimNode) startRPC() error { return errors.New("RPC already started") } - // add SimAdminAPI so that the network can call the AddPeer - // and RemovePeer RPC methods - apis := append(self.service.APIs(), rpc.API{ - Namespace: "admin", - Version: "1.0", - Service: &SimAdminAPI{self}, - }) + // add SimAdminAPI and PeerAPI so that the network can call the + // AddPeer, RemovePeer and PeerEvents RPC methods + apis := append(self.service.APIs(), []rpc.API{ + { + Namespace: "admin", + Version: "1.0", + Service: &SimAdminAPI{self}, + }, + { + Namespace: "eth", + Version: "1.0", + Service: &PeerAPI{func() p2p.Server { return self }}, + }, + }...) // start the RPC handler handler := rpc.NewServer() @@ -219,6 +228,10 @@ func (self *SimNode) AddPeer(node *discover.Node) { self.RunProtocol(na.(*SimNode), rw, rrw, peer) } +func (self *SimNode) SubscribePeers(ch chan *p2p.PeerEvent) event.Subscription { + return self.peerFeed.Subscribe(ch) +} + func (self *SimNode) PeerCount() int { self.lock.Lock() defer self.lock.Unlock() @@ -243,13 +256,13 @@ func (self *SimNode) RunProtocol(node *SimNode, rw, rrw p2p.MsgReadWriter, peer log.Trace(fmt.Sprintf("protocol starting on peer %v (connection with %v)", self.Id, id)) p := p2p.NewPeer(id.NodeID, id.Label(), []p2p.Cap{}) go func() { - self.network.DidConnect(self.Id, id) + self.peerFeed.Send(&p2p.PeerEvent{Type: p2p.PeerEventTypeAdd, Peer: id.NodeID}) err := protocol.Run(p, rw) <-peer.Readyc self.RemovePeer(node.Node()) peer.Errc <- err log.Trace(fmt.Sprintf("protocol quit on peer %v (connection with %v broken: %v)", self.Id, id, err)) - self.network.DidDisconnect(self.Id, id) + self.peerFeed.Send(&p2p.PeerEvent{Type: p2p.PeerEventTypeDrop, Peer: id.NodeID}) }() } diff --git a/p2p/adapters/types.go b/p2p/adapters/types.go index 47041f73b9..688dcfe53f 100644 --- a/p2p/adapters/types.go +++ b/p2p/adapters/types.go @@ -17,7 +17,6 @@ package adapters import ( - "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/rpc" @@ -79,22 +78,3 @@ type Reporter interface { DidConnect(*NodeId, *NodeId) error DidDisconnect(*NodeId, *NodeId) error } - -func RandomNodeId() *NodeId { - key, err := crypto.GenerateKey() - if err != nil { - panic("unable to generate key") - } - var id discover.NodeID - pubkey := crypto.FromECDSAPub(&key.PublicKey) - copy(id[:], pubkey[1:]) - return &NodeId{id} -} - -func RandomNodeIds(n int) []*NodeId { - var ids []*NodeId - for i := 0; i < n; i++ { - ids = append(ids, RandomNodeId()) - } - return ids -} diff --git a/p2p/peer.go b/p2p/peer.go index a9c20189a4..49126343c1 100644 --- a/p2p/peer.go +++ b/p2p/peer.go @@ -60,6 +60,26 @@ type protoHandshake struct { Rest []rlp.RawValue `rlp:"tail"` } +// PeerEventType is the type of peer events emitted by a p2p.Server +type PeerEventType string + +const ( + // PeerEventTypeAdd is the type of event emitted when a peer is added + // to a p2p.Server + PeerEventTypeAdd PeerEventType = "add" + + // PeerEventTypeDrop is the type of event emitted when a peer is + // dropped from a p2p.Server + PeerEventTypeDrop PeerEventType = "drop" +) + +// PeerEvent is an event emitted when peers are either added or dropped from +// a p2p.Server +type PeerEvent struct { + Type PeerEventType + Peer discover.NodeID +} + // Peer represents a connected remote node. type Peer struct { rw *conn diff --git a/p2p/protocols/protocol_test.go b/p2p/protocols/protocol_test.go index ac9bb0e53c..8a6380bb14 100644 --- a/p2p/protocols/protocol_test.go +++ b/p2p/protocols/protocol_test.go @@ -9,6 +9,7 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/adapters" + "github.com/ethereum/go-ethereum/p2p/simulations" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" ) @@ -117,8 +118,8 @@ func newProtocol(pp *p2ptest.TestPeerPool) adapters.ProtoCall { } func protocolTester(t *testing.T, pp *p2ptest.TestPeerPool) *p2ptest.ProtocolTester { - id := adapters.RandomNodeId() - return p2ptest.NewProtocolTester(t, id, 2, newProtocol(pp)) + conf := simulations.RandomNodeConfig() + return p2ptest.NewProtocolTester(t, conf.Id, 2, newProtocol(pp)) } func protoHandshakeExchange(id *adapters.NodeId, proto *protoHandshake) []p2ptest.Exchange { diff --git a/p2p/server.go b/p2p/server.go index 4b3cf1557d..bba0c54a08 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -27,6 +27,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/discv5" @@ -141,6 +142,7 @@ type Server interface { Stop() error AddPeer(node *discover.Node) RemovePeer(node *discover.Node) + SubscribePeers(ch chan *PeerEvent) event.Subscription PeerCount() int NodeInfo() *NodeInfo PeersInfo() []*PeerInfo @@ -180,6 +182,7 @@ type server struct { addpeer chan *conn delpeer chan peerDrop loopWG sync.WaitGroup // loop, listenLoop + peerFeed event.Feed } type peerOpFunc func(map[discover.NodeID]*Peer) @@ -305,6 +308,11 @@ func (srv *server) RemovePeer(node *discover.Node) { } } +// SubscribePeers subscribes the given channel to peer events +func (srv *server) SubscribePeers(ch chan *PeerEvent) event.Subscription { + return srv.peerFeed.Subscribe(ch) +} + // Self returns the local node's endpoint information. func (srv *server) Self() *discover.Node { srv.lock.Lock() @@ -770,6 +778,11 @@ func (srv *server) runPeer(p *Peer) { if srv.newPeerHook != nil { srv.newPeerHook(p) } + + // broadcast peer add / drop events + srv.peerFeed.Send(&PeerEvent{Type: PeerEventTypeAdd, Peer: p.ID()}) + defer srv.peerFeed.Send(&PeerEvent{Type: PeerEventTypeDrop, Peer: p.ID()}) + remoteRequested, err := p.run() // Note: run waits for existing peers to be sent on srv.delpeer // before returning, so this send should not select on srv.quit. diff --git a/p2p/simulations/mocker.go b/p2p/simulations/mocker.go index 0ad2b7cf3a..da25ff5533 100644 --- a/p2p/simulations/mocker.go +++ b/p2p/simulations/mocker.go @@ -163,7 +163,8 @@ func MockEvents(eventer *event.TypeMux, ids []*adapters.NodeId, conf *MockerConf var mustconnect []int for i := 0; len(offNodes) > 0 && i < nodesUp; i++ { c := rand.Intn(len(offNodes)) - sn := &Node{Id: offNodes[c]} + sn := &Node{} + sn.Id = offNodes[c] err := eventer.Post(sn.EmitEvent(ControlEvent)) if err != nil { panic(err.Error()) diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index 5e2bba1079..8a856f005a 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -27,14 +27,19 @@ package simulations import ( + "context" + "crypto/ecdsa" + "encoding/hex" "encoding/json" "fmt" "net/http" "reflect" "sync" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/adapters" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/rpc" @@ -176,9 +181,7 @@ func NewNodesController(net *Network) *ResourceController { &ResourceHandlers{ Create: &ResourceHandler{ Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { - nodeid := adapters.RandomNodeId() - net.NewNode(&NodeConfig{Id: nodeid}) - return &NodeConfig{Id: nodeid}, nil + return net.NewNode() }, }, Retrieve: &ResourceHandler{ @@ -289,7 +292,7 @@ func (self *Network) Subscribe(eventer *event.TypeMux, types ...interface{}) { func (self *Network) executeNodeEvent(ne *NodeControlEvent) { if ne.Up { - err := self.NewNode(&NodeConfig{Id: ne.Node.Id}) + err := self.NewNodeWithConfig(&ne.Node.NodeConfig) if err != nil { log.Trace(fmt.Sprintf("error execute event %v: %v", ne, err)) } @@ -370,9 +373,10 @@ type LiveEventer interface { } type Node struct { - Id *adapters.NodeId `json:"id"` - Up bool - config *NodeConfig + NodeConfig + + Up bool + na adapters.NodeAdapter controlFired bool } @@ -511,7 +515,55 @@ func (self *Node) Adapter() adapters.NodeAdapter { } type NodeConfig struct { - Id *adapters.NodeId `json:"Id"` + Id *adapters.NodeId + PrivateKey *ecdsa.PrivateKey +} + +type nodeConfigJSON struct { + Id string `json:"id"` + PrivateKey string `json:"private_key"` +} + +func (n *NodeConfig) MarshalJSON() ([]byte, error) { + return json.Marshal(nodeConfigJSON{ + n.Id.String(), + hex.EncodeToString(crypto.FromECDSA(n.PrivateKey)), + }) +} + +func (n *NodeConfig) UnmarshalJSON(data []byte) error { + var confJSON nodeConfigJSON + if err := json.Unmarshal(data, &confJSON); err != nil { + return err + } + + nodeID, err := discover.HexID(confJSON.Id) + if err != nil { + return err + } + n.Id = &adapters.NodeId{NodeID: nodeID} + + key, err := hex.DecodeString(confJSON.PrivateKey) + if err != nil { + return err + } + n.PrivateKey = crypto.ToECDSA(key) + + return nil +} + +func RandomNodeConfig() *NodeConfig { + key, err := crypto.GenerateKey() + if err != nil { + panic("unable to generate key") + } + var id discover.NodeID + pubkey := crypto.FromECDSAPub(&key.PublicKey) + copy(id[:], pubkey[1:]) + return &NodeConfig{ + Id: &adapters.NodeId{NodeID: id}, + PrivateKey: key, + } } // TODO: ignored for now @@ -530,9 +582,18 @@ type Know struct { // swap balance } -// NewNode adds a new node to the network +// NewNode adds a new node to the network with a random ID +func (self *Network) NewNode() (*NodeConfig, error) { + conf := RandomNodeConfig() + if err := self.NewNodeWithConfig(conf); err != nil { + return nil, err + } + return conf, nil +} + +// NewNodeWithConfig adds a new node to the network with the given config // errors if a node by the same id already exist -func (self *Network) NewNode(conf *NodeConfig) error { +func (self *Network) NewNodeWithConfig(conf *NodeConfig) error { self.lock.Lock() defer self.lock.Unlock() id := conf.Id @@ -544,9 +605,8 @@ func (self *Network) NewNode(conf *NodeConfig) error { self.nodeMap[id.NodeID] = len(self.Nodes) na := self.naf(conf) node := &Node{ - Id: conf.Id, - config: conf, - na: na, + NodeConfig: *conf, + na: na, } self.Nodes = append(self.Nodes, node) log.Trace(fmt.Sprintf("node %v created", id)) @@ -603,9 +663,49 @@ func (self *Network) Start(id *adapters.NodeId) error { log.Info(fmt.Sprintf("started node %v: %v", id, node.Up)) self.events.Post(node.EmitEvent(ControlEvent)) + + // subscribe to peer events + client, err := node.Adapter().Client() + if err != nil { + return fmt.Errorf("error getting rpc client for node %v: %s", id, err) + } + events := make(chan *p2p.PeerEvent) + sub, err := client.EthSubscribe(context.Background(), events, "peerEvents") + if err != nil { + return fmt.Errorf("error getting peer events for node %v: %s", id, err) + } + go self.watchPeerEvents(id, events, sub) return nil } +func (self *Network) watchPeerEvents(id *adapters.NodeId, events chan *p2p.PeerEvent, sub event.Subscription) { + defer sub.Unsubscribe() + for { + select { + case event, ok := <-events: + if !ok { + return + } + peer := &adapters.NodeId{NodeID: event.Peer} + switch event.Type { + case p2p.PeerEventTypeAdd: + if err := self.DidConnect(id, peer); err != nil { + log.Error(fmt.Sprintf("error generating connection up event %s => %s", id.Label(), peer.Label()), "err", err) + } + case p2p.PeerEventTypeDrop: + if err := self.DidDisconnect(id, peer); err != nil { + log.Error(fmt.Sprintf("error generating connection down event %s => %s", id.Label(), peer.Label()), "err", err) + } + } + case err := <-sub.Err(): + if err != nil { + log.Error(fmt.Sprintf("error getting peer events for node %v", id), "err", err) + } + return + } + } +} + // Stop(id) shuts down the node (relevant only for instance with own p2p or remote) func (self *Network) Stop(id *adapters.NodeId) error { node := self.GetNode(id) @@ -629,6 +729,7 @@ func (self *Network) Stop(id *adapters.NodeId) error { // calling the node's nodadapters Connect method // connection is established (as if) the first node dials out to the other func (self *Network) Connect(oneId, otherId *adapters.NodeId) error { + log.Debug(fmt.Sprintf("connecting %s to %s", oneId, otherId)) conn, err := self.GetOrCreateConn(oneId, otherId) if err != nil { return err @@ -700,7 +801,7 @@ func (self *Network) Disconnect(oneId, otherId *adapters.NodeId) error { } func (self *Network) DidConnect(one, other *adapters.NodeId) error { - conn, err := self.GetOrCreateConn(one, other) + conn, err := self.GetOrCreateConn(one, other) if err != nil { return fmt.Errorf("connection between %v and %v does not exist", one, other) } @@ -715,7 +816,7 @@ func (self *Network) DidConnect(one, other *adapters.NodeId) error { } func (self *Network) DidDisconnect(one, other *adapters.NodeId) error { - conn, err := self.GetOrCreateConn(one, other) + conn, err := self.GetOrCreateConn(one, other) if err != nil { return fmt.Errorf("connection between %v and %v does not exist", one, other) } @@ -808,6 +909,7 @@ func (self *Network) getConn(oneId, otherId *adapters.NodeId) *Conn { func (self *Network) Shutdown() { // disconnect all nodes for _, conn := range self.Conns { + log.Debug(fmt.Sprintf("disconnecting %s from %s", conn.One.Label(), conn.Other.Label())) if err := self.Disconnect(conn.One, conn.Other); err != nil { log.Warn(fmt.Sprintf("error disconnecting %s from %s", conn.One.Label(), conn.Other.Label()), "err", err) } @@ -815,6 +917,7 @@ func (self *Network) Shutdown() { // stop all nodes for _, node := range self.Nodes { + log.Debug(fmt.Sprintf("stopping node %s", node.Id.Label())) if err := node.na.Stop(); err != nil { log.Warn(fmt.Sprintf("error stopping node %s", node.Id.Label()), "err", err) } diff --git a/p2p/testing/protocoltester.go b/p2p/testing/protocoltester.go index e900fd6672..25970bf627 100644 --- a/p2p/testing/protocoltester.go +++ b/p2p/testing/protocoltester.go @@ -28,7 +28,7 @@ func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(*p2p.P } net.SetNaf(naf) - if err := net.NewNode(&simulations.NodeConfig{Id: id}); err != nil { + if err := net.NewNodeWithConfig(&simulations.NodeConfig{Id: id}); err != nil { panic(err.Error()) } if err := net.Start(id); err != nil { @@ -36,30 +36,35 @@ func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(*p2p.P } node := net.GetNodeAdapter(id).(*adapters.SimNode) - ids := adapters.RandomNodeIds(n) - ps := NewProtocolSession(node, ids) + peers := make([]*simulations.NodeConfig, n) + peerIDs := make([]*adapters.NodeId, n) + for i := 0; i < n; i++ { + peers[i] = simulations.RandomNodeConfig() + peerIDs[i] = peers[i].Id + } + ps := NewProtocolSession(node, peerIDs) self := &ProtocolTester{ ProtocolSession: ps, network: net, } - self.Connect(id, ids...) + self.Connect(id, peers...) return self } -func (self *ProtocolTester) Connect(selfId *adapters.NodeId, ids ...*adapters.NodeId) { - for _, id := range ids { - log.Trace(fmt.Sprintf("start node %v", id)) - if err := self.network.NewNode(&simulations.NodeConfig{Id: id}); err != nil { - panic(fmt.Sprintf("error starting peer %v: %v", id, err)) +func (self *ProtocolTester) Connect(selfId *adapters.NodeId, peers ...*simulations.NodeConfig) { + for _, peer := range peers { + log.Trace(fmt.Sprintf("start node %v", peer.Id)) + if err := self.network.NewNodeWithConfig(peer); err != nil { + panic(fmt.Sprintf("error starting peer %v: %v", peer.Id, err)) } - if err := self.network.Start(id); err != nil { - panic(fmt.Sprintf("error starting peer %v: %v", id, err)) + if err := self.network.Start(peer.Id); err != nil { + panic(fmt.Sprintf("error starting peer %v: %v", peer.Id, err)) } - log.Trace(fmt.Sprintf("connect to %v", id)) - if err := self.network.Connect(selfId, id); err != nil { - panic(fmt.Sprintf("error connecting to peer %v: %v", id, err)) + log.Trace(fmt.Sprintf("connect to %v", peer.Id)) + if err := self.network.Connect(selfId, peer.Id); err != nil { + panic(fmt.Sprintf("error connecting to peer %v: %v", peer.Id, err)) } } diff --git a/swarm/network/hive_test.go b/swarm/network/hive_test.go index 741e5c4432..6122ee061e 100644 --- a/swarm/network/hive_test.go +++ b/swarm/network/hive_test.go @@ -82,7 +82,6 @@ func TestRegisterAndConnect(t *testing.T) { }, ticker: make(chan time.Time), } - //pp.Start(tc.connect, tc.ping) pp.Start(s.TestNodeAdapter, tc.ping) defer pp.Stop() tc.ticker <- time.Now() diff --git a/swarm/network/pss_test.go b/swarm/network/pss_test.go index cebab91774..c3cd3b6e4d 100644 --- a/swarm/network/pss_test.go +++ b/swarm/network/pss_test.go @@ -86,6 +86,7 @@ func (n *pssTestNode) triggerCheck() { //go func() { n.trigger <- n.id }() go func() { n.trigger <- adapters.NewNodeId(n.Addr()) }() } + /* func (n *pssTestNode) RunProtocol(id *adapters.NodeId, rw, rrw p2p.MsgReadWriter, peer *adapters.Peer) error { return n.NodeAdapter.(adapters.ProtocolRunner).RunProtocol(id, rw, rrw, peer) @@ -275,7 +276,7 @@ func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes in }) testpeers := make(map[*adapters.NodeId]*pssTestPeer) nodes := newPssSimulationTester(t, numnodes, numfullnodes, net, trigger, vct, protocolName, protocolVersion, testpeers) - + ids := []*adapters.NodeId{} // connect the peers @@ -661,14 +662,14 @@ func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, net *s psss := make(map[*adapters.NodeId]*Pss) net.SetNaf(func(conf *simulations.NodeConfig) adapters.NodeAdapter { node := &pssTestNode{ - Pss: psss[conf.Id], - Hive: nil, - SimNode: &adapters.SimNode{}, - id: conf.Id, - network: net, - trigger: trigger, - ct: vct, - expectC: make(chan []int), + Pss: psss[conf.Id], + Hive: nil, + SimNode: &adapters.SimNode{}, + id: conf.Id, + network: net, + trigger: trigger, + ct: vct, + expectC: make(chan []int), } var handlefunc func(interface{}) error addr := NewPeerAddrFromNodeId(conf.Id) @@ -681,16 +682,18 @@ func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, net *s //node = newPssTester(t, psss[conf.Id], addr, 0, handlefunc, net, trigger) testservice := newPssTestService(t, handlefunc, node) nodes[conf.Id] = testservice.node - svc := adapters.NewSimNode(conf.Id, testservice, net) + svc := adapters.NewSimNode(conf.Id, testservice, net) testservice.Start(svc) //svcs[conf.Id] = testservice return node.SimNode }) - ids := adapters.RandomNodeIds(numnodes) - i := 0 - for _, id := range ids { - addr := NewPeerAddrFromNodeId(id) - psss[id] = makePss(addr.OverlayAddr()) + configs := make([]*simulations.NodeConfig, numnodes) + for i := 0; i < numnodes; i++ { + configs[i] = simulations.RandomNodeConfig() + } + for i, conf := range configs { + addr := NewPeerAddrFromNodeId(conf.Id) + psss[conf.Id] = makePss(addr.OverlayAddr()) if i < numfullnodes { tp := &pssTestPeer{ Peer: &protocols.Peer{ @@ -699,25 +702,24 @@ func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, net *s successC: make(chan bool), resultC: make(chan int), } - testpeers[id] = tp - targetprotocol := makeCustomProtocol(name, version, vct, testpeers[id]) - pssprotocol := NewPssProtocol(psss[id], &topic, vct, targetprotocol) - psss[id].Register(topic, pssprotocol.GetHandler()) + testpeers[conf.Id] = tp + targetprotocol := makeCustomProtocol(name, version, vct, testpeers[conf.Id]) + pssprotocol := NewPssProtocol(psss[conf.Id], &topic, vct, targetprotocol) + psss[conf.Id].Register(topic, pssprotocol.GetHandler()) } - net.NewNode(&simulations.NodeConfig{Id: id}) - if err := net.Start(id); err != nil { - t.Fatalf("error starting node %s: %s", id.Label(), err) + net.NewNodeWithConfig(conf) + if err := net.Start(conf.Id); err != nil { + t.Fatalf("error starting node %s: %s", conf.Id.Label(), err) } - i++ } return nodes //return svcs } -type pssTestService struct{ - node *pssTestNode // get addrs from this +type pssTestService struct { + node *pssTestNode // get addrs from this msgFunc func(interface{}) error } @@ -728,7 +730,7 @@ func newPssTestService(t *testing.T, handlefunc func(interface{}) error, testnod return &pssTestService{ //nid := adapters.NewNodeId(addr.UnderlayAddr()) msgFunc: handlefunc, - node: testnode, + node: testnode, } } @@ -749,17 +751,17 @@ func (self *pssTestService) Protocols() []p2p.Protocol { ct.Register(&PssMsg{}) //Bzz(addr.OverlayAddr(), addr.UnderlayAddr(), ct, srv, nil, nil).Run /* - node := &pssTestNode{ - Hive: hive, - Pss: ps, - NodeAdapter: nil, - id: nid, - network: net, - trigger: trigger, - ct: ct, - expectC: make(chan []int), - }*/ - + node := &pssTestNode{ + Hive: hive, + Pss: ps, + NodeAdapter: nil, + id: nid, + network: net, + trigger: trigger, + ct: ct, + expectC: make(chan []int), + }*/ + srv := func(p Peer) error { p.Register(&PssMsg{}, self.msgFunc) self.node.Add(p) @@ -768,9 +770,9 @@ func (self *pssTestService) Protocols() []p2p.Protocol { }) return nil } - + proto := Bzz(self.node.OverlayAddr(), self.node.UnderlayAddr(), ct, srv, nil, nil) - + return []p2p.Protocol{*proto} } @@ -781,11 +783,11 @@ func (self *pssTestService) APIs() []rpc.API { /* func newPssTester(t *testing.T, ps *Pss, addr *peerAddr, numsimnodes int, handlefunc func(interface{}) error, net *simulations.Network, trigger chan *adapters.NodeId) *pssTestNode { - + // set up the outer protocol - + srv := func(p Peer) error { p.Register(&PssMsg{}, handlefunc) @@ -795,13 +797,13 @@ func newPssTester(t *testing.T, ps *Pss, addr *peerAddr, numsimnodes int, handle }) return nil } - + node.run = Bzz(addr.OverlayAddr(), addr.UnderlayAddr(), ct, srv, nil, nil).Run nodeAdapter.Run = node.run - + node.NodeAdapter = adapters.NewSimNode(nid, net) - - + + return node } */ diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index 5957d07e55..2aa58e4a6c 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -54,7 +54,7 @@ func TestDiscoverySimulationDockerAdapter(t *testing.T) { }) net.SetNaf(func(conf *simulations.NodeConfig) adapters.NodeAdapter { - node, err := adapters.NewDockerNode(conf.Id, serviceName) + node, err := adapters.NewDockerNode(conf.Id, conf.PrivateKey, serviceName) if err != nil { panic(err) } @@ -84,7 +84,7 @@ func TestDiscoverySimulationExecAdapter(t *testing.T) { }) net.SetNaf(func(conf *simulations.NodeConfig) adapters.NodeAdapter { - node, err := adapters.NewExecNode(conf.Id, serviceName, baseDir) + node, err := adapters.NewExecNode(conf.Id, conf.PrivateKey, serviceName, baseDir) if err != nil { panic(err) } @@ -118,12 +118,16 @@ func testDiscoverySimulation(t *testing.T, setup func(net *simulations.Network, }) defer net.Shutdown() setup(net, trigger) - ids := adapters.RandomNodeIds(nodeCount) - for _, id := range ids { - net.NewNode(&simulations.NodeConfig{Id: id}) - if err := net.Start(id); err != nil { - t.Fatalf("error starting node %s: %s", id.Label(), err) + ids := make([]*adapters.NodeId, nodeCount) + for i := 0; i < nodeCount; i++ { + conf, err := net.NewNode() + if err != nil { + t.Fatalf("error starting node %s: %s", conf.Id.Label(), err) } + if err := net.Start(conf.Id); err != nil { + t.Fatalf("error starting node %s: %s", conf.Id.Label(), err) + } + ids[i] = conf.Id } // run a simulation which connects the 10 nodes in a ring and waits diff --git a/swarm/network/simulations/overlay.go b/swarm/network/simulations/overlay.go index 9cdea71102..889c59ead4 100644 --- a/swarm/network/simulations/overlay.go +++ b/swarm/network/simulations/overlay.go @@ -118,11 +118,16 @@ func nethook(conf *simulations.NetworkConfig) (simulations.NetworkControl, *simu conf.Backend = true net := NewNetwork(simulations.NewNetwork(conf)) - //ids := p2ptest.RandomNodeIds(10) - ids := adapters.RandomNodeIds(10) + ids := make([]*adapters.NodeId, 10) + for i := 0; i < 10; i++ { + conf, err := net.NewNode() + if err != nil { + panic(err.Error()) + } + ids[i] = conf.Id + } for i, id := range ids { - net.NewNode(&simulations.NodeConfig{Id: id}) var peerId *adapters.NodeId if i == 0 { peerId = ids[len(ids)-1] @@ -138,7 +143,6 @@ func nethook(conf *simulations.NetworkConfig) (simulations.NetworkControl, *simu for _, id := range ids { n := rand.Intn(1000) time.Sleep(time.Duration(n) * time.Millisecond) - net.NewNode(&simulations.NodeConfig{Id: id}) net.Start(id) log.Debug(fmt.Sprintf("node %v starting up", id)) // time.Sleep(1000 * time.Millisecond)