p2p/simulations: Support inter-node RPC connections

Signed-off-by: Lewis Marshall <lewis@lmars.net>

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,
+		ID:      config.ID,
-		Dir:    dir,
+		Dir:     dir,
-		Config: conf,
+		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
+	adapter *ExecAdapter
-	wsAddr string
+	client  *rpc.Client
-	newCmd func() *exec.Cmd
+	wsAddr  string
-	key    *ecdsa.PrivateKey
+	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
+	// initialize the devp2p stack
-	stack, err := startP2PNode(&conf.Stack, services)
+	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")
+}

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) {
+		return func(nodeCtx *node.ServiceContext) (node.Service, error) {
-			var snapshot []byte
+			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
 		}

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

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 {
+		"ping-pong": func(ctx *adapters.ServiceContext) (node.Service, error) {
-			return newPingPongService(id)
+			return newPingPongService(ctx.Config.ID), nil
 		},
 	}
 

p2p/simulations/http_test.go

@@ -24,10 +24,10 @@ type testService struct {
 	state atomic.Value
 }
 
-func newTestService(id discover.NodeID, snapshot []byte) node.Service {
+func newTestService(ctx *adapters.ServiceContext) (node.Service, error) {
-	svc := &testService{id: id}
+	svc := &testService{id: ctx.Config.ID}
-	svc.state.Store(snapshot)
+	svc.state.Store(ctx.Snapshot)
-	return svc
+	return svc, nil
 }
 
 func (t *testService) Protocols() []p2p.Protocol {

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 {
+		"test": func(ctx *adapters.ServiceContext) (node.Service, error) {
-			return &testNode{run}
+			return &testNode{run}, nil
 		},
-		"mock": func(id discover.NodeID, _ []byte) node.Service {
+		"mock": func(ctx *adapters.ServiceContext) (node.Service, error) {
-			return newMockNode()
+			return newMockNode(), nil
 		},
 	}
 	adapter := adapters.NewSimAdapter(services)

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 {
+func newService(ctx *adapters.ServiceContext) (node.Service, error) {
-	addr := network.NewAddrFromNodeID(id)
+	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
 }

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 {

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, err
-				return nil
 			}
 			dpa, err := storage.NewLocalDPA(cachedir)
 			if err != nil {
-				log.Error("local dpa creation failed", "error", err)
+				return nil, err
-				return nil
 			}
 
 			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)
+				return nil, err
-				os.Exit(1)
 			}
 
-			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
 		},
 	}
 }