mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-27 15:16:43 +00:00
Merge pull request #95 from ethersphere/network-testing-framework-services
p2p/simulations: Fix starting nodes with multiple services
This commit is contained in:
Viktor Trón
GitHub
commit
bdf4ba039c
9 changed files with 201 additions and 210 deletions
|
|
@ -43,8 +43,13 @@ func (d *DockerAdapter) Name() string {
|
|||
|
||||
// NewNode returns a new DockerNode using the given config
|
||||
func (d *DockerAdapter) NewNode(config *NodeConfig) (Node, error) {
|
||||
if _, exists := serviceFuncs[config.Service]; !exists {
|
||||
return nil, fmt.Errorf("unknown node service %q", config.Service)
|
||||
if len(config.Services) == 0 {
|
||||
return nil, errors.New("node must have at least one service")
|
||||
}
|
||||
for _, service := range config.Services {
|
||||
if _, exists := serviceFuncs[service]; !exists {
|
||||
return nil, fmt.Errorf("unknown node service %q", service)
|
||||
}
|
||||
}
|
||||
|
||||
// generate the config
|
||||
|
|
@ -76,22 +81,18 @@ type DockerNode struct {
|
|||
// 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 and _P2P_NODE_KEY
|
||||
// environment variables to the container using the --env flag.
|
||||
// It uses a shell so that we can pass the _P2P_NODE_CONFIG environment
|
||||
// variable 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}" --env _P2P_NODE_KEY="${_P2P_NODE_KEY}" %s p2p-node %s %s`,
|
||||
dockerImage, strings.Join(n.Services, " "), n.ID.String(),
|
||||
`exec docker run --interactive --env _P2P_NODE_CONFIG="${_P2P_NODE_CONFIG}" %s p2p-node %s %s`,
|
||||
dockerImage, strings.Join(n.Config.Node.Services, ","), n.ID.String(),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
func (n *DockerNode) GetService(name string) node.Service {
|
||||
return nil
|
||||
}
|
||||
|
||||
// dockerImage is the name of the docker image
|
||||
const dockerImage = "p2p-node"
|
||||
|
||||
|
|
|
|||
|
|
@ -46,8 +46,13 @@ func (e *ExecAdapter) Name() string {
|
|||
|
||||
// NewNode returns a new ExecNode using the given config
|
||||
func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
|
||||
if _, exists := serviceFuncs[config.Service]; !exists {
|
||||
return nil, fmt.Errorf("unknown node service %q", config.Service)
|
||||
if len(config.Services) == 0 {
|
||||
return nil, errors.New("node must have at least one service")
|
||||
}
|
||||
for _, service := range config.Services {
|
||||
if _, exists := serviceFuncs[service]; !exists {
|
||||
return nil, fmt.Errorf("unknown node service %q", service)
|
||||
}
|
||||
}
|
||||
|
||||
// create the node directory using the first 12 characters of the ID
|
||||
|
|
@ -88,12 +93,11 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
|
|||
// (so for example we can run the node in a remote Docker container and
|
||||
// still communicate with it).
|
||||
type ExecNode struct {
|
||||
ID *NodeId
|
||||
Dir string
|
||||
Config *execNodeConfig
|
||||
Cmd *exec.Cmd
|
||||
Info *p2p.NodeInfo
|
||||
Services []string
|
||||
ID *NodeId
|
||||
Dir string
|
||||
Config *execNodeConfig
|
||||
Cmd *exec.Cmd
|
||||
Info *p2p.NodeInfo
|
||||
|
||||
client *rpc.Client
|
||||
rpcMux *rpcMux
|
||||
|
|
@ -118,7 +122,7 @@ func (n *ExecNode) Client() (*rpc.Client, error) {
|
|||
// 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
|
||||
func (n *ExecNode) Start(snapshot []byte) (err error) {
|
||||
func (n *ExecNode) Start(snapshots map[string][]byte) (err error) {
|
||||
if n.Cmd != nil {
|
||||
return errors.New("already started")
|
||||
}
|
||||
|
|
@ -131,7 +135,7 @@ func (n *ExecNode) Start(snapshot []byte) (err error) {
|
|||
|
||||
// encode a copy of the config containing the snapshot
|
||||
confCopy := *n.Config
|
||||
confCopy.Snapshot = snapshot
|
||||
confCopy.Snapshots = snapshots
|
||||
confData, err := json.Marshal(confCopy)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error generating node config: %s", err)
|
||||
|
|
@ -165,17 +169,13 @@ func (n *ExecNode) Start(snapshot []byte) (err error) {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (n *ExecNode) GetService(name string) node.Service {
|
||||
return nil
|
||||
}
|
||||
|
||||
// execCommand returns a command which runs the node locally by exec'ing
|
||||
// the current binary but setting argv[0] to "p2p-node" so that the child
|
||||
// runs execP2PNode
|
||||
func (n *ExecNode) execCommand() *exec.Cmd {
|
||||
return &exec.Cmd{
|
||||
Path: reexec.Self(),
|
||||
Args: []string{"p2p-node", n.Services[0], n.ID.String()},
|
||||
Args: []string{"p2p-node", strings.Join(n.Config.Node.Services, ","), n.ID.String()},
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -231,14 +231,14 @@ func (n *ExecNode) ServeRPC(conn net.Conn) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// Snapshot creates a snapshot of the service state by calling the
|
||||
// Snapshots creates snapshots of the services by calling the
|
||||
// simulation_snapshot RPC method
|
||||
func (n *ExecNode) Snapshot() ([]byte, error) {
|
||||
func (n *ExecNode) Snapshots() (map[string][]byte, error) {
|
||||
if n.client == nil {
|
||||
return nil, errors.New("RPC not started")
|
||||
}
|
||||
var snapshot []byte
|
||||
return snapshot, n.client.Call(&snapshot, "simulation_snapshot")
|
||||
var snapshots map[string][]byte
|
||||
return snapshots, n.client.Call(&snapshots, "simulation_snapshot")
|
||||
}
|
||||
|
||||
func init() {
|
||||
|
|
@ -250,9 +250,9 @@ func init() {
|
|||
// execNodeConfig is used to serialize the node configuration so it can be
|
||||
// passed to the child process as a JSON encoded environment variable
|
||||
type execNodeConfig struct {
|
||||
Stack node.Config `json:"stack"`
|
||||
Node *NodeConfig `json:"node"`
|
||||
Snapshot []byte `json:"snapshot,omitempty"`
|
||||
Stack node.Config `json:"stack"`
|
||||
Node *NodeConfig `json:"node"`
|
||||
Snapshots map[string][]byte `json:"snapshot,omitempty"`
|
||||
}
|
||||
|
||||
// execP2PNode starts a devp2p node when the current binary is executed with
|
||||
|
|
@ -263,8 +263,8 @@ func execP2PNode() {
|
|||
glogger.Verbosity(log.LvlInfo)
|
||||
log.Root().SetHandler(glogger)
|
||||
|
||||
// read the service and ID from argv
|
||||
serviceName := os.Args[1]
|
||||
// read the services and ID from argv
|
||||
serviceNames := strings.Split(os.Args[1], ",")
|
||||
id := NewNodeIdFromHex(os.Args[2])
|
||||
|
||||
// decode the config
|
||||
|
|
@ -278,12 +278,19 @@ func execP2PNode() {
|
|||
}
|
||||
conf.Stack.P2P.PrivateKey = conf.Node.PrivateKey
|
||||
|
||||
// initialize the service
|
||||
serviceFunc, exists := serviceFuncs[serviceName]
|
||||
if !exists {
|
||||
log.Crit(fmt.Sprintf("unknown node service %q", serviceName))
|
||||
// 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)
|
||||
}
|
||||
services := serviceFunc(id, conf.Snapshot)
|
||||
|
||||
// use explicit IP address in ListenAddr so that Enode URL is usable
|
||||
if strings.HasPrefix(conf.Stack.P2P.ListenAddr, ":") {
|
||||
|
|
@ -326,52 +333,75 @@ func execP2PNode() {
|
|||
stack.Wait()
|
||||
}
|
||||
|
||||
func startP2PNode(conf *node.Config, services []node.Service) (*node.Node, error) {
|
||||
func startP2PNode(conf *node.Config, services map[string]node.Service) (*node.Node, error) {
|
||||
stack, err := node.New(conf)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, svc := range services {
|
||||
constructor := func(ctx *node.ServiceContext) (node.Service, error) {
|
||||
return &snapshotService{svc}, nil
|
||||
constructor := func(service node.Service) func(ctx *node.ServiceContext) (node.Service, error) {
|
||||
return func(ctx *node.ServiceContext) (node.Service, error) {
|
||||
return service, nil
|
||||
}
|
||||
if err := stack.Register(constructor); err != 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 wraps a node.Service and injects a snapshot API into the
|
||||
// list of RPC APIs
|
||||
// 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 {
|
||||
node.Service
|
||||
services map[string]node.Service
|
||||
}
|
||||
|
||||
func (s *snapshotService) APIs() []rpc.API {
|
||||
return append([]rpc.API{{
|
||||
return []rpc.API{{
|
||||
Namespace: "simulation",
|
||||
Version: "1.0",
|
||||
Service: SnapshotAPI{s.Service},
|
||||
}}, s.Service.APIs()...)
|
||||
Service: SnapshotAPI{s.services},
|
||||
}}
|
||||
}
|
||||
|
||||
// SnapshotAPI provides an RPC method to create a snapshot of a node.Service
|
||||
func (s *snapshotService) Protocols() []p2p.Protocol {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *snapshotService) Start(*p2p.Server) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *snapshotService) Stop() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// SnapshotAPI provides an RPC method to create snapshots of services
|
||||
type SnapshotAPI struct {
|
||||
service node.Service
|
||||
services map[string]node.Service
|
||||
}
|
||||
|
||||
func (api SnapshotAPI) Snapshot() ([]byte, error) {
|
||||
if s, ok := api.service.(interface {
|
||||
Snapshot() ([]byte, error)
|
||||
}); ok {
|
||||
return s.Snapshot()
|
||||
func (api SnapshotAPI) Snapshot() (map[string][]byte, error) {
|
||||
snapshots := make(map[string][]byte)
|
||||
for name, service := range api.services {
|
||||
if s, ok := service.(interface {
|
||||
Snapshot() ([]byte, error)
|
||||
}); ok {
|
||||
snap, err := s.Snapshot()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
snapshots[name] = snap
|
||||
}
|
||||
}
|
||||
return nil, nil
|
||||
return snapshots, nil
|
||||
}
|
||||
|
||||
// stdioConn wraps os.Stdin / os.Stdout with a no-op Close method so we can
|
||||
|
|
|
|||
|
|
@ -21,11 +21,9 @@ import (
|
|||
"fmt"
|
||||
"math"
|
||||
"net"
|
||||
"reflect"
|
||||
"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"
|
||||
"github.com/ethereum/go-ethereum/p2p/discover"
|
||||
|
|
@ -35,8 +33,9 @@ import (
|
|||
// SimAdapter is a NodeAdapter which creates in-memory nodes and connects them
|
||||
// using an in-memory p2p.MsgReadWriter pipe
|
||||
type SimAdapter struct {
|
||||
mtx sync.RWMutex
|
||||
nodes map[discover.NodeID]*SimNode
|
||||
mtx sync.RWMutex
|
||||
nodes map[discover.NodeID]*SimNode
|
||||
services map[string]ServiceFunc
|
||||
}
|
||||
|
||||
// NewSimAdapter creates a SimAdapter which is capable of running in-memory
|
||||
|
|
@ -44,7 +43,8 @@ type SimAdapter struct {
|
|||
// node is passed to the NewNode function in the NodeConfig)
|
||||
func NewSimAdapter(services map[string]ServiceFunc) *SimAdapter {
|
||||
return &SimAdapter{
|
||||
nodes: make(map[discover.NodeID]*SimNode),
|
||||
nodes: make(map[discover.NodeID]*SimNode),
|
||||
services: services,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -55,8 +55,6 @@ func (s *SimAdapter) Name() string {
|
|||
|
||||
// NewNode returns a new SimNode using the given config
|
||||
func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
|
||||
var nodeprotos []p2p.Protocol
|
||||
|
||||
s.mtx.Lock()
|
||||
defer s.mtx.Unlock()
|
||||
|
||||
|
|
@ -66,55 +64,23 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
|
|||
return nil, fmt.Errorf("node already exists: %s", id)
|
||||
}
|
||||
|
||||
// check the service is valid and initialize it
|
||||
/*
|
||||
serviceFunc, exists := s.services[config.Service]
|
||||
if !exists {
|
||||
return nil, fmt.Errorf("unknown node service %q", config.Service)
|
||||
}
|
||||
|
||||
node := &SimNode{
|
||||
Id: id,
|
||||
config: config,
|
||||
adapter: s,
|
||||
serviceFunc: serviceFunc,
|
||||
*/
|
||||
//serviceFunc, exists := s.services[config.Service]
|
||||
|
||||
//if !exists {
|
||||
// return nil, fmt.Errorf("unknown node service %q", config.Service)
|
||||
//}
|
||||
//service := serviceFunc(id)
|
||||
|
||||
_, err := node.New(&node.Config{
|
||||
P2P: p2p.Config{
|
||||
PrivateKey: config.PrivateKey,
|
||||
MaxPeers: math.MaxInt32,
|
||||
NoDiscovery: true,
|
||||
Protocols: nodeprotos,
|
||||
Dialer: s,
|
||||
EnableMsgEvents: true,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
// check the services are valid
|
||||
if len(config.Services) == 0 {
|
||||
return nil, errors.New("node must have at least one service")
|
||||
}
|
||||
|
||||
for _, service := range serviceFuncs[config.Service](id, nil) {
|
||||
for _, proto := range service.Protocols() {
|
||||
nodeprotos = append(nodeprotos, proto)
|
||||
for _, service := range config.Services {
|
||||
if _, exists := s.services[service]; !exists {
|
||||
return nil, fmt.Errorf("unknown node service %q", service)
|
||||
}
|
||||
}
|
||||
|
||||
simnode := &SimNode{
|
||||
Id: id,
|
||||
serviceFunc: serviceFuncs[config.Service],
|
||||
adapter: s,
|
||||
config: config,
|
||||
running: []node.Service{},
|
||||
node := &SimNode{
|
||||
Id: id,
|
||||
config: config,
|
||||
adapter: s,
|
||||
}
|
||||
s.nodes[id.NodeID] = simnode
|
||||
return simnode, nil
|
||||
s.nodes[id.NodeID] = node
|
||||
return node, nil
|
||||
}
|
||||
|
||||
func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) {
|
||||
|
|
@ -146,15 +112,14 @@ func (s *SimAdapter) GetNode(id discover.NodeID) (*SimNode, bool) {
|
|||
// It implements the p2p.Server interface so it can be used transparently
|
||||
// by the underlying service.
|
||||
type SimNode struct {
|
||||
lock sync.RWMutex
|
||||
Id *NodeId
|
||||
config *NodeConfig
|
||||
adapter *SimAdapter
|
||||
serviceFunc ServiceFunc
|
||||
node *node.Node
|
||||
client *rpc.Client
|
||||
rpcMux *rpcMux
|
||||
running []node.Service
|
||||
lock sync.RWMutex
|
||||
Id *NodeId
|
||||
config *NodeConfig
|
||||
adapter *SimAdapter
|
||||
node *node.Node
|
||||
running []node.Service
|
||||
client *rpc.Client
|
||||
rpcMux *rpcMux
|
||||
}
|
||||
|
||||
// Addr returns the node's discovery address
|
||||
|
|
@ -191,38 +156,37 @@ func (self *SimNode) ServeRPC(conn net.Conn) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// Snapshot creates a snapshot of the service state by calling the
|
||||
// Snapshots creates snapshots of the services by calling the
|
||||
// simulation_snapshot RPC method
|
||||
func (self *SimNode) Snapshot() ([]byte, error) {
|
||||
func (self *SimNode) Snapshots() (map[string][]byte, error) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
if self.client == nil {
|
||||
return nil, errors.New("RPC not started")
|
||||
}
|
||||
var snapshot []byte
|
||||
return snapshot, self.client.Call(&snapshot, "simulation_snapshot")
|
||||
var snapshots map[string][]byte
|
||||
return snapshots, self.client.Call(&snapshots, "simulation_snapshot")
|
||||
}
|
||||
|
||||
// Start starts the RPC handler and the underlying service
|
||||
func (self *SimNode) Start(snapshot []byte) error {
|
||||
func (self *SimNode) Start(snapshots map[string][]byte) error {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
if self.node != nil {
|
||||
return errors.New("node already started")
|
||||
}
|
||||
|
||||
services := []node.ServiceConstructor{}
|
||||
|
||||
sf := self.serviceFunc(self.Id, snapshot)
|
||||
|
||||
for i, _ := range sf {
|
||||
service := sf[i]
|
||||
sc := func(ctx *node.ServiceContext) (node.Service, error) {
|
||||
newService := func(name string) func(ctx *node.ServiceContext) (node.Service, error) {
|
||||
return func(ctx *node.ServiceContext) (node.Service, error) {
|
||||
var snapshot []byte
|
||||
if snapshots != nil {
|
||||
snapshot = snapshots[name]
|
||||
}
|
||||
serviceFunc := self.adapter.services[name]
|
||||
service := serviceFunc(self.Id, snapshot)
|
||||
self.running = append(self.running, service)
|
||||
return service, nil
|
||||
}
|
||||
log.Debug(fmt.Sprintf("servicefunc yield: %v %p %p", reflect.TypeOf(sf[i]), sf[i], sc))
|
||||
services = append(services, sc)
|
||||
self.running = append(self.running, sf[i])
|
||||
}
|
||||
|
||||
node, err := node.New(&node.Config{
|
||||
|
|
@ -239,9 +203,8 @@ func (self *SimNode) Start(snapshot []byte) error {
|
|||
return err
|
||||
}
|
||||
|
||||
for _, service := range services {
|
||||
log.Debug(fmt.Sprintf("service %v", service))
|
||||
if err := node.Register(service); err != nil {
|
||||
for _, name := range self.config.Services {
|
||||
if err := node.Register(newService(name)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
|
@ -281,17 +244,11 @@ func (self *SimNode) Stop() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// Service returns the underlying running node.Service matching the supplied service type
|
||||
func (self *SimNode) Service(servicetype interface{}) node.Service {
|
||||
// Services returns the underlying services
|
||||
func (self *SimNode) Services() []node.Service {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
typ := reflect.TypeOf(servicetype)
|
||||
for _, service := range self.running {
|
||||
if reflect.TypeOf(service) == typ {
|
||||
return service
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return self.running
|
||||
}
|
||||
|
||||
func (self *SimNode) Server() *p2p.Server {
|
||||
|
|
|
|||
|
|
@ -52,7 +52,7 @@ type Node interface {
|
|||
ServeRPC(net.Conn) error
|
||||
|
||||
// Start starts the node with the given snapshot
|
||||
Start(snapshot []byte) error
|
||||
Start(snapshots map[string][]byte) error
|
||||
|
||||
// Stop stops the node
|
||||
Stop() error
|
||||
|
|
@ -60,8 +60,8 @@ type Node interface {
|
|||
// NodeInfo returns information about the node
|
||||
NodeInfo() *p2p.NodeInfo
|
||||
|
||||
// Snapshot creates a snapshot of the running service
|
||||
Snapshot() ([]byte, error)
|
||||
// Snapshots creates snapshots of the running services
|
||||
Snapshots() (map[string][]byte, error)
|
||||
}
|
||||
|
||||
// NodeAdapter is an object which creates Nodes to be used in a simulation
|
||||
|
|
@ -127,26 +127,26 @@ type NodeConfig struct {
|
|||
// Name is a human friendly name for the node like "node01"
|
||||
Name string
|
||||
|
||||
// Service is the name of the services which should be run when starting
|
||||
// Services are the names of the services which should be run when starting
|
||||
// the node (for SimNodes it should be the names of services contained
|
||||
// in SimAdapter.services, for other nodes it should be services
|
||||
// registered by calling the RegisterService function)
|
||||
Service string
|
||||
Services []string
|
||||
}
|
||||
|
||||
// nodeConfigJSON is used to encode and decode NodeConfig as JSON by converting
|
||||
// all fields to strings
|
||||
type nodeConfigJSON struct {
|
||||
Id string `json:"id"`
|
||||
PrivateKey string `json:"private_key"`
|
||||
Name string `json:"name"`
|
||||
Service string `json:"service"`
|
||||
Id string `json:"id"`
|
||||
PrivateKey string `json:"private_key"`
|
||||
Name string `json:"name"`
|
||||
Services []string `json:"services"`
|
||||
}
|
||||
|
||||
func (n *NodeConfig) MarshalJSON() ([]byte, error) {
|
||||
confJSON := nodeConfigJSON{
|
||||
Name: n.Name,
|
||||
Service: n.Service,
|
||||
Name: n.Name,
|
||||
Services: n.Services,
|
||||
}
|
||||
if n.Id != nil {
|
||||
confJSON.Id = n.Id.String()
|
||||
|
|
@ -180,7 +180,7 @@ func (n *NodeConfig) UnmarshalJSON(data []byte) error {
|
|||
}
|
||||
|
||||
n.Name = confJSON.Name
|
||||
n.Service = confJSON.Service
|
||||
n.Services = confJSON.Services
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
@ -202,18 +202,17 @@ func RandomNodeConfig() *NodeConfig {
|
|||
}
|
||||
|
||||
// Services is a collection of services which can be run in a simulation
|
||||
// it is mapped to strings representing TYPES of nodes
|
||||
type Services map[string]ServiceFunc
|
||||
|
||||
// ServiceFunc returns a node.Service which can be used to boot devp2p nodes
|
||||
type ServiceFunc func(id *NodeId, snapshot []byte) []node.Service
|
||||
type ServiceFunc func(id *NodeId, snapshot []byte) node.Service
|
||||
|
||||
// serviceFuncs is a map of registered services which are used to boot devp2p
|
||||
// nodes
|
||||
var serviceFuncs = make(Services)
|
||||
|
||||
// RegisterServices registers the given ServiceFuncs which can then be used to
|
||||
// start devp2p nodes
|
||||
// start devp2p nodes using either the Exec or Docker adapters
|
||||
func RegisterServices(services Services) {
|
||||
for name, f := range services {
|
||||
if _, exists := serviceFuncs[name]; exists {
|
||||
|
|
|
|||
|
|
@ -190,7 +190,7 @@ func (self *Msg) String() string {
|
|||
// NewNode adds a new node to the network with a random ID
|
||||
func (self *Network) NewNode() (*Node, error) {
|
||||
conf := adapters.RandomNodeConfig()
|
||||
conf.Service = self.DefaultService
|
||||
conf.Services = []string{self.DefaultService}
|
||||
return self.NewNodeWithConfig(conf)
|
||||
}
|
||||
|
||||
|
|
@ -203,8 +203,8 @@ func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error)
|
|||
if conf.Name == "" {
|
||||
conf.Name = fmt.Sprintf("node%02d", len(self.Nodes)+1)
|
||||
}
|
||||
if conf.Service == "" {
|
||||
conf.Service = self.DefaultService
|
||||
if len(conf.Services) == 0 {
|
||||
conf.Services = []string{self.DefaultService}
|
||||
}
|
||||
|
||||
_, found := self.nodeMap[id.NodeID]
|
||||
|
|
@ -286,10 +286,10 @@ func (self *Network) StopAll() error {
|
|||
|
||||
// Start(id) starts up the node (relevant only for instance with own p2p or remote)
|
||||
func (self *Network) Start(id *adapters.NodeId) error {
|
||||
return self.startWithSnapshot(id, nil)
|
||||
return self.startWithSnapshots(id, nil)
|
||||
}
|
||||
|
||||
func (self *Network) startWithSnapshot(id *adapters.NodeId, snapshot []byte) error {
|
||||
func (self *Network) startWithSnapshots(id *adapters.NodeId, snapshots map[string][]byte) error {
|
||||
node := self.GetNode(id)
|
||||
if node == nil {
|
||||
return fmt.Errorf("node %v does not exist", id)
|
||||
|
|
@ -298,7 +298,7 @@ func (self *Network) startWithSnapshot(id *adapters.NodeId, snapshot []byte) err
|
|||
return fmt.Errorf("node %v already up", id)
|
||||
}
|
||||
log.Trace(fmt.Sprintf("starting node %v: %v using %v", id, node.Up, self.nodeAdapter.Name()))
|
||||
if err := node.Start(snapshot); err != nil {
|
||||
if err := node.Start(snapshots); err != nil {
|
||||
log.Warn(fmt.Sprintf("start up failed: %v", err))
|
||||
return err
|
||||
}
|
||||
|
|
@ -625,8 +625,8 @@ type Snapshot struct {
|
|||
type NodeSnapshot struct {
|
||||
Node
|
||||
|
||||
// Snapshot is arbitrary data gathered from calling node.Snapshot()
|
||||
Snapshot []byte `json:"snapshot,omitempty"`
|
||||
// Snapshot is arbitrary data gathered from calling node.Snapshots()
|
||||
Snapshots map[string][]byte `json:"snapshots,omitempty"`
|
||||
}
|
||||
|
||||
// Snapshot creates a network snapshot
|
||||
|
|
@ -642,11 +642,11 @@ func (self *Network) Snapshot() (*Snapshot, error) {
|
|||
if !node.Up {
|
||||
continue
|
||||
}
|
||||
snapshot, err := node.Snapshot()
|
||||
snapshots, err := node.Snapshots()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
snap.Nodes[i].Snapshot = snapshot
|
||||
snap.Nodes[i].Snapshots = snapshots
|
||||
}
|
||||
for i, conn := range self.Conns {
|
||||
snap.Conns[i] = *conn
|
||||
|
|
@ -662,7 +662,7 @@ func (self *Network) Load(snap *Snapshot) error {
|
|||
if !node.Up {
|
||||
continue
|
||||
}
|
||||
if err := self.startWithSnapshot(node.Config.Id, node.Snapshot); err != nil {
|
||||
if err := self.startWithSnapshots(node.Config.Id, node.Snapshots); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ func (self *ProtocolSession) trigger(trig Trigger) error {
|
|||
if !ok {
|
||||
return fmt.Errorf("trigger: peer %v does not exist (1- %v)", trig.Peer, len(self.Ids))
|
||||
}
|
||||
mockNode, ok := simNode.Service(&mockNode{}).(*mockNode)
|
||||
mockNode, ok := simNode.Services()[0].(*mockNode)
|
||||
if !ok {
|
||||
return fmt.Errorf("trigger: peer %v is not a mock", trig.Peer)
|
||||
}
|
||||
|
|
@ -92,7 +92,7 @@ func (self *ProtocolSession) expect(exp Expect) error {
|
|||
if !ok {
|
||||
return fmt.Errorf("trigger: peer %v does not exist (1- %v)", exp.Peer, len(self.Ids))
|
||||
}
|
||||
mockNode, ok := simNode.Service(&mockNode{}).(*mockNode)
|
||||
mockNode, ok := simNode.Services()[0].(*mockNode)
|
||||
if !ok {
|
||||
return fmt.Errorf("trigger: peer %v is not a mock", exp.Peer)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,18 +19,17 @@ type ProtocolTester struct {
|
|||
}
|
||||
|
||||
func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(*p2p.Peer, p2p.MsgReadWriter) error) *ProtocolTester {
|
||||
//func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(*p2p.Peer, p2p.MsgReadWriter) error) *ProtocolTester {
|
||||
services := adapters.Services{
|
||||
"test": func(id *adapters.NodeId, _ []byte) []node.Service {
|
||||
return []node.Service{&testNode{run}}
|
||||
"test": func(id *adapters.NodeId, _ []byte) node.Service {
|
||||
return &testNode{run}
|
||||
},
|
||||
"mock": func(id *adapters.NodeId, _ []byte) []node.Service {
|
||||
return []node.Service{newMockNode()}
|
||||
"mock": func(id *adapters.NodeId, _ []byte) node.Service {
|
||||
return newMockNode()
|
||||
},
|
||||
}
|
||||
adapter := adapters.NewSimAdapter(services)
|
||||
net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{})
|
||||
if _, err := net.NewNodeWithConfig(&adapters.NodeConfig{Id: id, Service: "test"}); err != nil {
|
||||
if _, err := net.NewNodeWithConfig(&adapters.NodeConfig{Id: id, Services: []string{"test"}}); err != nil {
|
||||
panic(err.Error())
|
||||
}
|
||||
if err := net.Start(id); err != nil {
|
||||
|
|
@ -42,7 +41,7 @@ func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(*p2p.P
|
|||
peerIDs := make([]*adapters.NodeId, n)
|
||||
for i := 0; i < n; i++ {
|
||||
peers[i] = adapters.RandomNodeConfig()
|
||||
peers[i].Service = "mock"
|
||||
peers[i].Services = []string{"mock"}
|
||||
peerIDs[i] = peers[i].Id
|
||||
}
|
||||
events := make(chan *p2p.PeerEvent, 1000)
|
||||
|
|
|
|||
|
|
@ -33,7 +33,7 @@ func NewSimulation() *Simulation {
|
|||
}
|
||||
}
|
||||
|
||||
func (s *Simulation) NewService(id *adapters.NodeId, snapshot []byte) []node.Service {
|
||||
func (s *Simulation) NewService(id *adapters.NodeId, snapshot []byte) node.Service {
|
||||
s.mtx.Lock()
|
||||
store, ok := s.stores[id.NodeID]
|
||||
if !ok {
|
||||
|
|
@ -64,7 +64,7 @@ func (s *Simulation) NewService(id *adapters.NodeId, snapshot []byte) []node.Ser
|
|||
HiveParams: hp,
|
||||
}
|
||||
|
||||
return []node.Service{network.NewBzz(config, kad, store)}
|
||||
return network.NewBzz(config, kad, store)
|
||||
}
|
||||
|
||||
func createMockers() map[string]*simulations.MockerConfig {
|
||||
|
|
@ -124,7 +124,7 @@ func setupMocker(net *simulations.Network) []*adapters.NodeId {
|
|||
defer close(ch)
|
||||
ch <- network.NewAddrFromNodeId(peerId)
|
||||
}()
|
||||
if err := net.GetNode(id).Node.(*adapters.SimNode).Service(&network.Bzz{}).(*network.Bzz).Hive.Register(ch); err != nil {
|
||||
if err := net.GetNode(id).Node.(*adapters.SimNode).Services()[0].(*network.Bzz).Hive.Register(ch); err != nil {
|
||||
panic(err.Error())
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -242,8 +242,7 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int
|
|||
|
||||
nodeCount := 5
|
||||
net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
|
||||
Id: "0",
|
||||
DefaultService: "psstest",
|
||||
Id: "0",
|
||||
})
|
||||
defer net.Shutdown()
|
||||
|
||||
|
|
@ -254,7 +253,7 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int
|
|||
|
||||
for i := 0; i < nodeCount; i++ {
|
||||
nodeconfig := adapters.RandomNodeConfig()
|
||||
nodeconfig.Service = "psstest"
|
||||
nodeconfig.Services = []string{"bzz", "pss"}
|
||||
node, err := net.NewNodeWithConfig(nodeconfig)
|
||||
if err != nil {
|
||||
t.Fatalf("error starting node: %s", err)
|
||||
|
|
@ -436,28 +435,25 @@ func triggerChecks(trigger chan *adapters.NodeId, net *simulations.Network, id *
|
|||
}
|
||||
|
||||
func newServices() adapters.Services {
|
||||
|
||||
stateStore := adapters.NewSimStateStore()
|
||||
kademlias := make(map[*adapters.NodeId]*network.Kademlia)
|
||||
kademlia := func(id *adapters.NodeId) *network.Kademlia {
|
||||
if k, ok := kademlias[id]; ok {
|
||||
return k
|
||||
}
|
||||
addr := network.NewAddrFromNodeId(id)
|
||||
params := network.NewKadParams()
|
||||
params.MinProxBinSize = 2
|
||||
params.MaxBinSize = 3
|
||||
params.MinBinSize = 1
|
||||
params.MaxRetries = 1000
|
||||
params.RetryExponent = 2
|
||||
params.RetryInterval = 1000000
|
||||
kademlias[id] = network.NewKademlia(addr.Over(), params)
|
||||
return kademlias[id]
|
||||
}
|
||||
return adapters.Services{
|
||||
"psstest": func(id *adapters.NodeId, snapshot []byte) []node.Service {
|
||||
addr := network.NewAddrFromNodeId(id)
|
||||
|
||||
kadparams := network.NewKadParams()
|
||||
kadparams.MinProxBinSize = 2
|
||||
kadparams.MaxBinSize = 3
|
||||
kadparams.MinBinSize = 1
|
||||
kadparams.MaxRetries = 1000
|
||||
kadparams.RetryExponent = 2
|
||||
kadparams.RetryInterval = 1000000
|
||||
kademlia := network.NewKademlia(addr.Over(), kadparams)
|
||||
|
||||
config := &network.BzzConfig{
|
||||
OverlayAddr: addr.Over(),
|
||||
UnderlayAddr: addr.Under(),
|
||||
HiveParams: network.NewHiveParams(),
|
||||
}
|
||||
|
||||
config.HiveParams.KeepAliveInterval = time.Second
|
||||
|
||||
"pss": func(id *adapters.NodeId, snapshot []byte) node.Service {
|
||||
cachedir, err := ioutil.TempDir("", "pss-cache")
|
||||
if err != nil {
|
||||
log.Error("create pss cache tmpdir failed", "error", err)
|
||||
|
|
@ -470,7 +466,7 @@ func newServices() adapters.Services {
|
|||
}
|
||||
|
||||
pssp := NewPssParams()
|
||||
ps := NewPss(kademlia, dpa, pssp)
|
||||
ps := NewPss(kademlia(id), dpa, pssp)
|
||||
|
||||
ping := &pssPing{
|
||||
quitC: make(chan struct{}),
|
||||
|
|
@ -481,7 +477,16 @@ func newServices() adapters.Services {
|
|||
os.Exit(1)
|
||||
}
|
||||
|
||||
return []node.Service{network.NewBzz(config, kademlia, adapters.NewSimStateStore()), ps}
|
||||
return ps
|
||||
},
|
||||
"bzz": func(id *adapters.NodeId, snapshot []byte) node.Service {
|
||||
addr := network.NewAddrFromNodeId(id)
|
||||
config := &network.BzzConfig{
|
||||
OverlayAddr: addr.Over(),
|
||||
UnderlayAddr: addr.Under(),
|
||||
HiveParams: network.NewHiveParams(),
|
||||
}
|
||||
return network.NewBzz(config, kademlia(id), stateStore)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue