diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go index 61cd5c5c01..319e4353a2 100644 --- a/p2p/simulations/adapters/exec.go +++ b/p2p/simulations/adapters/exec.go @@ -36,12 +36,17 @@ import ( // execP2PNode function for more information. type ExecAdapter struct { BaseDir string + + nodes map[discover.NodeID]*ExecNode } // NewExecAdapter returns an ExecAdapter which stores node data in // subdirectories of the given base directory func NewExecAdapter(baseDir string) *ExecAdapter { - return &ExecAdapter{BaseDir: baseDir} + return &ExecAdapter{ + BaseDir: baseDir, + nodes: make(map[discover.NodeID]*ExecNode), + } } // Name returns the name of the adapter for logging purpoeses @@ -86,11 +91,13 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) { conf.Stack.P2P.ListenAddr = "127.0.0.1:0" node := &ExecNode{ - ID: config.ID, - Dir: dir, - Config: conf, + ID: config.ID, + Dir: dir, + Config: conf, + adapter: e, } node.newCmd = node.execCommand + e.nodes[node.ID] = node return node, nil } @@ -108,10 +115,11 @@ type ExecNode struct { Cmd *exec.Cmd Info *p2p.NodeInfo - client *rpc.Client - wsAddr string - newCmd func() *exec.Cmd - key *ecdsa.PrivateKey + adapter *ExecAdapter + client *rpc.Client + wsAddr string + newCmd func() *exec.Cmd + key *ecdsa.PrivateKey } // Addr returns the node's enode URL @@ -149,6 +157,10 @@ func (n *ExecNode) Start(snapshots map[string][]byte) (err error) { // encode a copy of the config containing the snapshot confCopy := *n.Config confCopy.Snapshots = snapshots + confCopy.PeerAddrs = make(map[string]string) + for id, node := range n.adapter.nodes { + confCopy.PeerAddrs[id.String()] = node.wsAddr + } confData, err := json.Marshal(confCopy) if err != nil { return fmt.Errorf("error generating node config: %s", err) @@ -315,7 +327,8 @@ func init() { type execNodeConfig struct { Stack node.Config `json:"stack"` Node *NodeConfig `json:"node"` - Snapshots map[string][]byte `json:"snapshot,omitempty"` + Snapshots map[string][]byte `json:"snapshots,omitempty"` + PeerAddrs map[string]string `json:"peer_addrs,omitempty"` } // execP2PNode starts a devp2p node when the current binary is executed with @@ -326,9 +339,8 @@ func execP2PNode() { glogger.Verbosity(log.LvlInfo) log.Root().SetHandler(glogger) - // read the services and ID from argv + // read the services from argv serviceNames := strings.Split(os.Args[1], ",") - id := discover.MustHexID(os.Args[2]) // decode the config confEnv := os.Getenv("_P2P_NODE_CONFIG") @@ -341,20 +353,6 @@ func execP2PNode() { } conf.Stack.P2P.PrivateKey = conf.Node.PrivateKey - // initialize the services - services := make(map[string]node.Service, len(serviceNames)) - for _, name := range serviceNames { - serviceFunc, exists := serviceFuncs[name] - if !exists { - log.Crit(fmt.Sprintf("unknown node service %q", name)) - } - var snapshot []byte - if conf.Snapshots != nil { - snapshot = conf.Snapshots[name] - } - services[name] = serviceFunc(id, snapshot) - } - // use explicit IP address in ListenAddr so that Enode URL is usable if strings.HasPrefix(conf.Stack.P2P.ListenAddr, ":") { addrs, err := net.InterfaceAddrs() @@ -369,12 +367,54 @@ func execP2PNode() { } } - // start the devp2p stack - stack, err := startP2PNode(&conf.Stack, services) + // initialize the devp2p stack + stack, err := node.New(&conf.Stack) if err != nil { - log.Crit("error starting p2p node", "err", err) + log.Crit("error creating node stack", "err", err) } + // register the services, collecting them into a map so we can wrap + // them in a snapshot service + services := make(map[string]node.Service, len(serviceNames)) + for _, name := range serviceNames { + serviceFunc, exists := serviceFuncs[name] + if !exists { + log.Crit("unknown node service", "name", name) + } + constructor := func(nodeCtx *node.ServiceContext) (node.Service, error) { + ctx := &ServiceContext{ + RPCDialer: &wsRPCDialer{addrs: conf.PeerAddrs}, + NodeContext: nodeCtx, + Config: conf.Node, + } + if conf.Snapshots != nil { + ctx.Snapshot = conf.Snapshots[name] + } + service, err := serviceFunc(ctx) + if err != nil { + return nil, err + } + services[name] = service + return service, nil + } + if err := stack.Register(constructor); err != nil { + log.Crit("error starting service", "name", name, "err", err) + } + } + + // register the snapshot service + if err := stack.Register(func(ctx *node.ServiceContext) (node.Service, error) { + return &snapshotService{services}, nil + }); err != nil { + log.Crit("error starting snapshot service", "err", err) + } + + // start the stack + if err := stack.Start(); err != nil { + log.Crit("error stating node stack", "err", err) + } + + // stop the stack if we get a SIGTERM signal go func() { sigc := make(chan os.Signal, 1) signal.Notify(sigc, syscall.SIGTERM) @@ -384,33 +424,10 @@ func execP2PNode() { stack.Stop() }() + // wait for the stack to exit stack.Wait() } -func startP2PNode(conf *node.Config, services map[string]node.Service) (*node.Node, error) { - stack, err := node.New(conf) - if err != nil { - return nil, err - } - constructor := func(service node.Service) func(ctx *node.ServiceContext) (node.Service, error) { - return func(ctx *node.ServiceContext) (node.Service, error) { - return service, nil - } - } - for _, service := range services { - if err := stack.Register(constructor(service)); err != nil { - return nil, err - } - } - if err := stack.Register(constructor(&snapshotService{services})); err != nil { - return nil, err - } - if err := stack.Start(); err != nil { - return nil, err - } - return stack, nil -} - // snapshotService is a node.Service which wraps a list of services and // exposes an API to generate a snapshot of those services type snapshotService struct { @@ -457,3 +474,15 @@ func (api SnapshotAPI) Snapshot() (map[string][]byte, error) { } return snapshots, nil } + +type wsRPCDialer struct { + addrs map[string]string +} + +func (w *wsRPCDialer) DialRPC(id discover.NodeID) (*rpc.Client, error) { + addr, ok := w.addrs[id.String()] + if !ok { + return nil, fmt.Errorf("unknown node: %s", id) + } + return rpc.DialWebsocket(context.Background(), addr, "http://localhost") +} diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index f0a989d829..cf3a404d73 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -97,6 +97,24 @@ func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) { return pipe2, nil } +func (s *SimAdapter) DialRPC(id discover.NodeID) (*rpc.Client, error) { + simNode, ok := s.GetNode(id) + if !ok { + return nil, fmt.Errorf("unknown node: %s", id) + } + simNode.lock.RLock() + node := simNode.node + simNode.lock.RUnlock() + if node == nil { + return nil, errors.New("node not started") + } + handler, err := node.RPCHandler() + if err != nil { + return nil, err + } + return rpc.DialInProc(handler), nil +} + // GetNode returns the node with the given ID if it exists func (s *SimAdapter) GetNode(id discover.NodeID) (*SimNode, bool) { s.mtx.RLock() @@ -180,13 +198,20 @@ func (self *SimNode) Start(snapshots map[string][]byte) error { } newService := func(name string) func(ctx *node.ServiceContext) (node.Service, error) { - return func(ctx *node.ServiceContext) (node.Service, error) { - var snapshot []byte + return func(nodeCtx *node.ServiceContext) (node.Service, error) { + ctx := &ServiceContext{ + RPCDialer: self.adapter, + NodeContext: nodeCtx, + Config: self.config, + } if snapshots != nil { - snapshot = snapshots[name] + ctx.Snapshot = snapshots[name] } serviceFunc := self.adapter.services[name] - service := serviceFunc(self.ID, snapshot) + service, err := serviceFunc(ctx) + if err != nil { + return nil, err + } self.running = append(self.running, service) return service, nil } diff --git a/p2p/simulations/adapters/types.go b/p2p/simulations/adapters/types.go index 034d162297..f5ac3ec4ac 100644 --- a/p2p/simulations/adapters/types.go +++ b/p2p/simulations/adapters/types.go @@ -158,11 +158,23 @@ func RandomNodeConfig() *NodeConfig { } } +type ServiceContext struct { + RPCDialer + + NodeContext *node.ServiceContext + Config *NodeConfig + Snapshot []byte +} + +type RPCDialer interface { + DialRPC(id discover.NodeID) (*rpc.Client, error) +} + // Services is a collection of services which can be run in a simulation type Services map[string]ServiceFunc // ServiceFunc returns a node.Service which can be used to boot devp2p nodes -type ServiceFunc func(id discover.NodeID, snapshot []byte) node.Service +type ServiceFunc func(ctx *ServiceContext) (node.Service, error) // serviceFuncs is a map of registered services which are used to boot devp2p // nodes diff --git a/p2p/simulations/examples/connectivity.go b/p2p/simulations/examples/connectivity.go index 6174c094e8..95b7c8be7e 100644 --- a/p2p/simulations/examples/connectivity.go +++ b/p2p/simulations/examples/connectivity.go @@ -27,8 +27,8 @@ func main() { log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false)))) services := map[string]adapters.ServiceFunc{ - "ping-pong": func(id discover.NodeID, snapshot []byte) node.Service { - return newPingPongService(id) + "ping-pong": func(ctx *adapters.ServiceContext) (node.Service, error) { + return newPingPongService(ctx.Config.ID), nil }, } diff --git a/p2p/simulations/http_test.go b/p2p/simulations/http_test.go index d91127450e..84ceb0f677 100644 --- a/p2p/simulations/http_test.go +++ b/p2p/simulations/http_test.go @@ -24,10 +24,10 @@ type testService struct { state atomic.Value } -func newTestService(id discover.NodeID, snapshot []byte) node.Service { - svc := &testService{id: id} - svc.state.Store(snapshot) - return svc +func newTestService(ctx *adapters.ServiceContext) (node.Service, error) { + svc := &testService{id: ctx.Config.ID} + svc.state.Store(ctx.Snapshot) + return svc, nil } func (t *testService) Protocols() []p2p.Protocol { diff --git a/p2p/testing/protocoltester.go b/p2p/testing/protocoltester.go index bbaa18a8ae..9e58a79d18 100644 --- a/p2p/testing/protocoltester.go +++ b/p2p/testing/protocoltester.go @@ -21,11 +21,11 @@ type ProtocolTester struct { func NewProtocolTester(t *testing.T, id discover.NodeID, n int, run func(*p2p.Peer, p2p.MsgReadWriter) error) *ProtocolTester { services := adapters.Services{ - "test": func(id discover.NodeID, _ []byte) node.Service { - return &testNode{run} + "test": func(ctx *adapters.ServiceContext) (node.Service, error) { + return &testNode{run}, nil }, - "mock": func(id discover.NodeID, _ []byte) node.Service { - return newMockNode() + "mock": func(ctx *adapters.ServiceContext) (node.Service, error) { + return newMockNode(), nil }, } adapter := adapters.NewSimAdapter(services) diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index 13d1f1fbef..fb038ed2a7 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -179,8 +179,8 @@ func triggerChecks(trigger chan discover.NodeID, net *simulations.Network, id di return nil } -func newService(id discover.NodeID, snapshot []byte) node.Service { - addr := network.NewAddrFromNodeID(id) +func newService(ctx *adapters.ServiceContext) (node.Service, error) { + addr := network.NewAddrFromNodeID(ctx.Config.ID) kp := network.NewKadParams() kp.MinProxBinSize = testMinProxBinSize @@ -200,5 +200,5 @@ func newService(id discover.NodeID, snapshot []byte) node.Service { HiveParams: hp, } - return network.NewBzz(config, kad, nil) + return network.NewBzz(config, kad, nil), nil } diff --git a/swarm/network/simulations/overlay.go b/swarm/network/simulations/overlay.go index dbf3ddb867..319eeec5c4 100644 --- a/swarm/network/simulations/overlay.go +++ b/swarm/network/simulations/overlay.go @@ -33,7 +33,8 @@ func NewSimulation() *Simulation { } } -func (s *Simulation) NewService(id discover.NodeID, snapshot []byte) node.Service { +func (s *Simulation) NewService(ctx *adapters.ServiceContext) (node.Service, error) { + id := ctx.Config.ID s.mtx.Lock() store, ok := s.stores[id] if !ok { @@ -64,7 +65,7 @@ func (s *Simulation) NewService(id discover.NodeID, snapshot []byte) node.Servic HiveParams: hp, } - return network.NewBzz(config, kad, store) + return network.NewBzz(config, kad, store), nil } func createMockers() map[string]*simulations.MockerConfig { diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index f540d6aa3e..3b400ade20 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -453,16 +453,15 @@ func newServices() adapters.Services { return kademlias[id] } return adapters.Services{ - "pss": func(id discover.NodeID, snapshot []byte) node.Service { + "pss": func(ctx *adapters.ServiceContext) (node.Service, error) { + id := ctx.Config.ID cachedir, err := ioutil.TempDir("", "pss-cache") if err != nil { - log.Error("create pss cache tmpdir failed", "error", err) - return nil + return nil, err } dpa, err := storage.NewLocalDPA(cachedir) if err != nil { - log.Error("local dpa creation failed", "error", err) - return nil + return nil, err } pssp := NewPssParams() @@ -473,20 +472,20 @@ func newServices() adapters.Services { } err = RegisterPssProtocol(ps, &pssPingTopic, pssPingProtocol, newPssPingProtocol(ping.pssPingHandler)) if err != nil { - log.Error("Couldnt register pss protocol", "err", err) - os.Exit(1) + return nil, err } - return ps + return ps, nil }, - "bzz": func(id discover.NodeID, snapshot []byte) node.Service { + "bzz": func(ctx *adapters.ServiceContext) (node.Service, error) { + id := ctx.Config.ID addr := network.NewAddrFromNodeID(id) config := &network.BzzConfig{ OverlayAddr: addr.Over(), UnderlayAddr: addr.Under(), HiveParams: network.NewHiveParams(), } - return network.NewBzz(config, kademlia(id), stateStore) + return network.NewBzz(config, kademlia(id), stateStore), nil }, } }