From d7eb6b5a70d4f49887ce982068007821dad8b407 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Wed, 10 May 2017 18:20:14 -0700 Subject: [PATCH] p2p/simulations: Snapshot support Signed-off-by: Lewis Marshall --- p2p/simulations/adapters/docker.go | 23 +- p2p/simulations/adapters/exec.go | 145 +++++----- p2p/simulations/adapters/inproc.go | 107 +++++--- p2p/simulations/adapters/types.go | 9 +- p2p/simulations/cmd/p2psim/main.go | 38 +++ p2p/simulations/events.go | 47 +++- p2p/simulations/examples/connectivity.go | 10 +- p2p/simulations/http.go | 55 +++- p2p/simulations/http_test.go | 247 +++++++++++++++--- p2p/simulations/journal.go | 235 ----------------- p2p/simulations/journal_test.go | 143 ---------- p2p/simulations/network.go | 79 +++++- .../simulations/discovery/discovery_test.go | 2 +- swarm/network/simulations/overlay.go | 6 +- 14 files changed, 592 insertions(+), 554 deletions(-) delete mode 100644 p2p/simulations/journal.go delete mode 100644 p2p/simulations/journal_test.go diff --git a/p2p/simulations/adapters/docker.go b/p2p/simulations/adapters/docker.go index bc801a8a94..a9460df0df 100644 --- a/p2p/simulations/adapters/docker.go +++ b/p2p/simulations/adapters/docker.go @@ -47,18 +47,19 @@ func (d *DockerAdapter) NewNode(config *NodeConfig) (Node, error) { } // generate the config - conf := node.DefaultConfig - conf.DataDir = "/data" - conf.P2P.EnableMsgEvents = true - conf.P2P.NoDiscovery = true - conf.P2P.NAT = nil + conf := &execNodeConfig{ + Stack: node.DefaultConfig, + Node: config, + } + conf.Stack.DataDir = "/data" + conf.Stack.P2P.EnableMsgEvents = true + conf.Stack.P2P.NoDiscovery = true + conf.Stack.P2P.NAT = nil node := &DockerNode{ ExecNode: ExecNode{ - ID: config.Id, - Service: config.Service, - Config: &conf, - key: config.PrivateKey, + ID: config.Id, + Config: conf, }, } node.newCmd = node.dockerCommand @@ -80,8 +81,8 @@ 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, n.Service, n.ID.String(), + `exec docker run --interactive --env _P2P_NODE_CONFIG="${_P2P_NODE_CONFIG}" %s p2p-node %s %s`, + dockerImage, n.Config.Node.Service, n.ID.String(), ), ) } diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go index bb34a14cf7..706cdcfd9f 100644 --- a/p2p/simulations/adapters/exec.go +++ b/p2p/simulations/adapters/exec.go @@ -3,7 +3,6 @@ package adapters import ( "context" "crypto/ecdsa" - "encoding/hex" "encoding/json" "errors" "fmt" @@ -18,7 +17,6 @@ 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" @@ -60,22 +58,23 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) { } // generate the config - conf := node.DefaultConfig - conf.DataDir = filepath.Join(dir, "data") - conf.P2P.EnableMsgEvents = true - conf.P2P.NoDiscovery = true - conf.P2P.NAT = nil + conf := &execNodeConfig{ + Stack: node.DefaultConfig, + Node: config, + } + conf.Stack.DataDir = filepath.Join(dir, "data") + conf.Stack.P2P.EnableMsgEvents = true + conf.Stack.P2P.NoDiscovery = true + conf.Stack.P2P.NAT = nil // listen on a random localhost port (we'll get the actual port after // starting the node through the RPC admin.nodeInfo method) - conf.P2P.ListenAddr = "127.0.0.1:0" + conf.Stack.P2P.ListenAddr = "127.0.0.1:0" node := &ExecNode{ - ID: config.Id, - Service: config.Service, - Dir: dir, - Config: &conf, - key: config.PrivateKey, + ID: config.Id, + Dir: dir, + Config: conf, } node.newCmd = node.execCommand return node, nil @@ -89,12 +88,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 - Service string - Dir string - Config *node.Config - Cmd *exec.Cmd - Info *p2p.NodeInfo + ID *NodeId + Dir string + Config *execNodeConfig + Cmd *exec.Cmd + Info *p2p.NodeInfo client *rpc.Client rpcMux *rpcMux @@ -116,11 +114,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, -// 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) { +// 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) { if n.Cmd != nil { return errors.New("already started") } @@ -131,15 +128,14 @@ func (n *ExecNode) Start() (err error) { } }() - // encode the config - conf, err := json.Marshal(n.Config) + // encode a copy of the config containing the snapshot + confCopy := *n.Config + confCopy.Snapshot = snapshot + confData, err := json.Marshal(confCopy) if err != nil { 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() @@ -148,10 +144,7 @@ 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), - fmt.Sprintf("_P2P_NODE_KEY=%s", key), - ) + cmd.Env = append(os.Environ(), fmt.Sprintf("_P2P_NODE_CONFIG=%s", confData)) if err := cmd.Start(); err != nil { return fmt.Errorf("error starting node: %s", err) } @@ -177,7 +170,7 @@ func (n *ExecNode) Start() (err error) { func (n *ExecNode) execCommand() *exec.Cmd { return &exec.Cmd{ Path: reexec.Self(), - Args: []string{"p2p-node", n.Service, n.ID.String()}, + Args: []string{"p2p-node", n.Config.Node.Service, n.ID.String()}, } } @@ -214,12 +207,11 @@ func (n *ExecNode) Stop() error { // NodeInfo returns information about the node func (n *ExecNode) NodeInfo() *p2p.NodeInfo { - if n.client == nil { - return n.Info + info := &p2p.NodeInfo{ + ID: n.ID.String(), } - info := &p2p.NodeInfo{} - if err := n.client.Call(&info, "admin_nodeInfo"); err != nil { - return n.Info + if n.client != nil { + n.client.Call(&info, "admin_nodeInfo") } return info } @@ -234,16 +226,33 @@ func (n *ExecNode) ServeRPC(conn net.Conn) error { return nil } +// Snapshot creates a snapshot of the service state by calling the +// simulation_snapshot RPC method +func (n *ExecNode) Snapshot() ([]byte, error) { + if n.client == nil { + return nil, errors.New("RPC not started") + } + var snapshot []byte + return snapshot, n.client.Call(&snapshot, "simulation_snapshot") +} + func init() { // register a reexec function to start a devp2p node when the current // binary is executed as "p2p-node" reexec.Register("p2p-node", execP2PNode) } +// 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"` +} + // execP2PNode starts a devp2p node when the current binary is executed with -// argv[0] being "p2p-node", reading the service / ID from argv[1] / argv[2], -// the node config from the _P2P_NODE_CONFIG environment variable and the -// private key from the _P2P_NODE_KEY environment variable +// argv[0] being "p2p-node", reading the service / ID from argv[1] / argv[2] +// and the node config from the _P2P_NODE_CONFIG environment variable func execP2PNode() { glogger := log.NewGlogHandler(log.StreamHandler(os.Stderr, log.LogfmtFormat())) glogger.Verbosity(log.LvlInfo) @@ -258,45 +267,35 @@ func execP2PNode() { if confEnv == "" { log.Crit("missing _P2P_NODE_CONFIG") } - var conf node.Config + var conf execNodeConfig if err := json.Unmarshal([]byte(confEnv), &conf); err != nil { 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) + 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)) } - service := serviceFunc(id) + service := serviceFunc(id, conf.Snapshot) // use explicit IP address in ListenAddr so that Enode URL is usable - if strings.HasPrefix(conf.P2P.ListenAddr, ":") { + if strings.HasPrefix(conf.Stack.P2P.ListenAddr, ":") { addrs, err := net.InterfaceAddrs() if err != nil { log.Crit("error getting IP address", "err", err) } for _, addr := range addrs { if ip, ok := addr.(*net.IPNet); ok && !ip.IP.IsLoopback() { - conf.P2P.ListenAddr = ip.IP.String() + conf.P2P.ListenAddr + conf.Stack.P2P.ListenAddr = ip.IP.String() + conf.Stack.P2P.ListenAddr break } } } // start the devp2p stack - stack, err := startP2PNode(&conf, service) + stack, err := startP2PNode(&conf.Stack, service) if err != nil { log.Crit("error starting p2p node", "err", err) } @@ -328,7 +327,7 @@ func startP2PNode(conf *node.Config, service node.Service) (*node.Node, error) { return nil, err } constructor := func(ctx *node.ServiceContext) (node.Service, error) { - return service, nil + return &snapshotService{service}, nil } if err := stack.Register(constructor); err != nil { return nil, err @@ -339,6 +338,34 @@ func startP2PNode(conf *node.Config, service node.Service) (*node.Node, error) { return stack, nil } +// snapshotService wraps a node.Service and injects a snapshot API into the +// list of RPC APIs +type snapshotService struct { + node.Service +} + +func (s *snapshotService) APIs() []rpc.API { + return append([]rpc.API{{ + Namespace: "simulation", + Version: "1.0", + Service: SnapshotAPI{s.Service}, + }}, s.Service.APIs()...) +} + +// SnapshotAPI provides an RPC method to create a snapshot of a node.Service +type SnapshotAPI struct { + service node.Service +} + +func (api SnapshotAPI) Snapshot() ([]byte, error) { + if s, ok := api.service.(interface { + Snapshot() ([]byte, error) + }); ok { + return s.Snapshot() + } + return nil, nil +} + // stdioConn wraps os.Stdin / os.Stdout with a no-op Close method so we can // use stdio for RPC messages type stdioConn struct { diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index 481fe18768..6f27607cf4 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -70,21 +70,13 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) { if !exists { return nil, fmt.Errorf("unknown node service %q", config.Service) } - service := serviceFunc(id) - - // for simplicity, only support single protocol services (simulating - // multiple protocols on the same peer is extra effort, and we don't - // currently run any simulations which run multiple protocols) - if len(service.Protocols()) != 1 { - return nil, errors.New("service must have a single protocol") - } node := &SimNode{ - Id: id, - adapter: s, - service: service, - peers: make(map[discover.NodeID]MsgReadWriteCloser), - dropPeers: make(chan struct{}), + Id: id, + adapter: s, + serviceFunc: serviceFunc, + peers: make(map[discover.NodeID]MsgReadWriteCloser), + dropPeers: make(chan struct{}), } s.nodes[id.NodeID] = node return node, nil @@ -113,14 +105,15 @@ type MsgReadWriteCloser interface { // It implements the p2p.Server interface so it can be used transparently // by the underlying service. type SimNode struct { - lock sync.RWMutex - Id *NodeId - adapter *SimAdapter - service node.Service - peers map[discover.NodeID]MsgReadWriteCloser - peerFeed event.Feed - client *rpc.Client - rpcMux *rpcMux + lock sync.RWMutex + Id *NodeId + adapter *SimAdapter + running node.Service + serviceFunc ServiceFunc + peers map[discover.NodeID]MsgReadWriteCloser + peerFeed event.Feed + client *rpc.Client + rpcMux *rpcMux // dropPeers is used to force peer disconnects when // the node is stopped @@ -161,13 +154,35 @@ func (self *SimNode) ServeRPC(conn net.Conn) error { return nil } -// Start starts the RPC handler and the underlying service -func (self *SimNode) Start() error { +// Start initializes the service, starts the RPC handler and then starts +// the service +func (self *SimNode) Start(snapshot []byte) error { + service := self.serviceFunc(self.Id, snapshot) + + // for simplicity, only support single protocol services (simulating + // multiple protocols on the same peer is extra effort, and we don't + // currently run any simulations which run multiple protocols) + if len(service.Protocols()) != 1 { + return errors.New("service must have a single protocol") + } + self.dropPeers = make(chan struct{}) - if err := self.startRPC(); err != nil { + if err := self.startRPC(service); err != nil { return err } - return self.service.Start(self) + self.running = service + return service.Start(&simServer{self}) +} + +// simServer wraps a SimNode but modifies the Start method signature so that +// it implements the p2p.Server interface (the Start method is never actually +// called when using the SimAdapter) +type simServer struct { + *SimNode +} + +func (s *simServer) Start() error { + return nil } // Stop stops the RPC handler, stops the underlying service and disconnects @@ -175,25 +190,24 @@ func (self *SimNode) Start() error { func (self *SimNode) Stop() error { self.stopRPC() close(self.dropPeers) - return self.service.Stop() + return self.running.Stop() } -// Running returns whether or not the service is running by checking if the -// RPC client is set +// Running returns whether or not the service is running func (self *SimNode) Running() bool { self.lock.Lock() defer self.lock.Unlock() - return self.client != nil + return self.running != nil } -// Service returns the underlying node.Service +// Service returns the running node.Service func (self *SimNode) Service() node.Service { - return self.service + return self.running } // startRPC starts an RPC server and connects to it using an in-process RPC // client -func (self *SimNode) startRPC() error { +func (self *SimNode) startRPC(service node.Service) error { self.lock.Lock() defer self.lock.Unlock() if self.client != nil { @@ -202,7 +216,7 @@ func (self *SimNode) startRPC() error { // add SimAdminAPI so that the network can call the // AddPeer, RemovePeer and PeerEvents RPC methods - apis := append(self.service.APIs(), []rpc.API{ + apis := append(service.APIs(), []rpc.API{ { Namespace: "admin", Version: "1.0", @@ -292,17 +306,21 @@ func (self *SimNode) PeerCount() int { // NodeInfo returns information about the node func (self *SimNode) NodeInfo() *p2p.NodeInfo { + self.lock.Lock() + defer self.lock.Unlock() info := &p2p.NodeInfo{ ID: self.Id.String(), Enode: self.Node().String(), Protocols: make(map[string]interface{}), } - for _, proto := range self.service.Protocols() { - nodeInfo := interface{}("unknown") - if query := proto.NodeInfo; query != nil { - nodeInfo = proto.NodeInfo() + if self.running != nil { + for _, proto := range self.running.Protocols() { + nodeInfo := interface{}("unknown") + if query := proto.NodeInfo; query != nil { + nodeInfo = proto.NodeInfo() + } + info.Protocols[proto.Name] = nodeInfo } - info.Protocols[proto.Name] = nodeInfo } return info } @@ -312,6 +330,17 @@ func (self *SimNode) PeersInfo() (info []*p2p.PeerInfo) { return nil } +// Snapshot creates a snapshot of the running service +func (self *SimNode) Snapshot() ([]byte, error) { + self.lock.Lock() + service := self.running + self.lock.Unlock() + if service == nil { + return nil, errors.New("service not running") + } + return SnapshotAPI{service}.Snapshot() +} + // RunProtocol runs the underlying service's protocol with the peer using the // given MsgReadWriteCloser, emitting peer add / drop events for peer event // subscribers @@ -325,7 +354,7 @@ func (self *SimNode) RunProtocol(peer *SimNode, rw MsgReadWriteCloser) { id := peer.Id log.Trace(fmt.Sprintf("protocol starting on peer %v (connection with %v)", self.Id, id)) - protocol := self.service.Protocols()[0] + protocol := self.running.Protocols()[0] p := p2p.NewPeer(id.NodeID, id.Label(), []p2p.Cap{}) go func() { // emit peer add event diff --git a/p2p/simulations/adapters/types.go b/p2p/simulations/adapters/types.go index 4bfdded042..e7f15ba8f7 100644 --- a/p2p/simulations/adapters/types.go +++ b/p2p/simulations/adapters/types.go @@ -51,14 +51,17 @@ type Node interface { // node's RPC multiplexer ServeRPC(net.Conn) error - // Start starts the node - Start() error + // Start starts the node with the given snapshot + Start(snapshot []byte) error // Stop stops the node Stop() error // NodeInfo returns information about the node NodeInfo() *p2p.NodeInfo + + // Snapshot creates a snapshot of the running service + Snapshot() ([]byte, error) } // NodeAdapter is an object which creates Nodes to be used in a simulation @@ -202,7 +205,7 @@ func RandomNodeConfig() *NodeConfig { type Services map[string]ServiceFunc // ServiceFunc returns a node.Service which can be used to boot devp2p nodes -type ServiceFunc func(id *NodeId) 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 diff --git a/p2p/simulations/cmd/p2psim/main.go b/p2p/simulations/cmd/p2psim/main.go index b0b53a5016..1de64abf57 100644 --- a/p2p/simulations/cmd/p2psim/main.go +++ b/p2p/simulations/cmd/p2psim/main.go @@ -91,6 +91,18 @@ func main() { Usage: "stream network events", Action: streamNetwork, }, + { + Name: "snapshot", + ArgsUsage: "", + Usage: "create a network snapshot to stdout", + Action: createSnapshot, + }, + { + Name: "load", + ArgsUsage: "", + Usage: "load a network snapshot from stdin", + Action: loadSnapshot, + }, }, }, { @@ -241,6 +253,32 @@ func streamNetwork(ctx *cli.Context) error { } } +func createSnapshot(ctx *cli.Context) error { + args := ctx.Args() + if len(args) != 1 { + return cli.ShowCommandHelp(ctx, ctx.Command.Name) + } + networkID := args[0] + snap, err := client.CreateSnapshot(networkID) + if err != nil { + return err + } + return json.NewEncoder(os.Stdout).Encode(snap) +} + +func loadSnapshot(ctx *cli.Context) error { + args := ctx.Args() + if len(args) != 1 { + return cli.ShowCommandHelp(ctx, ctx.Command.Name) + } + networkID := args[0] + snap := &simulations.Snapshot{} + if err := json.NewDecoder(os.Stdin).Decode(snap); err != nil { + return err + } + return client.LoadSnapshot(networkID, snap) +} + func listNodes(ctx *cli.Context) error { args := ctx.Args() if len(args) != 1 { diff --git a/p2p/simulations/events.go b/p2p/simulations/events.go index 8e8168c34b..4cd0f077eb 100644 --- a/p2p/simulations/events.go +++ b/p2p/simulations/events.go @@ -5,48 +5,79 @@ import ( "time" ) +// EventType is the type of event emitted by a simulation network type EventType string const ( + // EventTypeNode is the type of event emitted when a node is either + // created, started or stopped EventTypeNode EventType = "node" + + // EventTypeConn is the type of event emitted when a connection is + // is either established or dropped between two nodes EventTypeConn EventType = "conn" - EventTypeMsg EventType = "msg" + + // EventTypeMsg is the type of event emitted when a p2p message it + // sent between two nodes + EventTypeMsg EventType = "msg" ) +// Event is an event emitted by a simulation network type Event struct { - Type EventType `json:"type"` - Time time.Time `json:"time"` - Control bool `json:"control"` + // Type is the type of the event + Type EventType `json:"type"` + // Time is the time the event happened + Time time.Time `json:"time"` + + // Control indicates whether the event is the result of a controlled + // action in the network + Control bool `json:"control"` + + // Node is set if the type is EventTypeNode Node *Node `json:"node,omitempty"` + + // Conn is set if the type is EventTypeConn Conn *Conn `json:"conn,omitempty"` - Msg *Msg `json:"msg,omitempty"` + + // Msg is set if the type is EventTypeMsg + Msg *Msg `json:"msg,omitempty"` } +// NewEvent creates a new event for the given object which should be either a +// Node, Conn or Msg. +// +// The object is copied so that the event represents the state of the object +// when NewEvent is called. func NewEvent(v interface{}) *Event { event := &Event{Time: time.Now()} switch v := v.(type) { case *Node: event.Type = EventTypeNode - event.Node = v + node := *v + event.Node = &node case *Conn: event.Type = EventTypeConn - event.Conn = v + conn := *v + event.Conn = &conn case *Msg: event.Type = EventTypeMsg - event.Msg = v + msg := *v + event.Msg = &msg default: panic(fmt.Sprintf("invalid event type: %T", v)) } return event } +// ControlEvent creates a new control event func ControlEvent(v interface{}) *Event { event := NewEvent(v) event.Control = true return event } +// String returns the string representation of the event func (e *Event) String() string { switch e.Type { case EventTypeNode: diff --git a/p2p/simulations/examples/connectivity.go b/p2p/simulations/examples/connectivity.go index c25875cb71..43beff951a 100644 --- a/p2p/simulations/examples/connectivity.go +++ b/p2p/simulations/examples/connectivity.go @@ -26,7 +26,7 @@ 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 *adapters.NodeId) node.Service { + "ping-pong": func(id *adapters.NodeId, snapshot []byte) node.Service { return newPingPongService(id) }, } @@ -42,7 +42,7 @@ func main() { case "sim": log.Info("using sim adapter") - config.Adapter = adapters.NewSimAdapter(services) + config.NewAdapter = func() adapters.NodeAdapter { return adapters.NewSimAdapter(services) } case "exec": tmpdir, err := ioutil.TempDir("", "p2p-example") @@ -51,15 +51,15 @@ func main() { } defer os.RemoveAll(tmpdir) log.Info("using exec adapter", "tmpdir", tmpdir) - config.Adapter = adapters.NewExecAdapter(tmpdir) + config.NewAdapter = func() adapters.NodeAdapter { return adapters.NewExecAdapter(tmpdir) } case "docker": log.Info("using docker adapter") - var err error - config.Adapter, err = adapters.NewDockerAdapter() + adapter, err := adapters.NewDockerAdapter() if err != nil { log.Crit("error creating docker adapter", "err", err) } + config.NewAdapter = func() adapters.NodeAdapter { return adapter } default: log.Crit(fmt.Sprintf("unknown node adapter %q", *adapter)) diff --git a/p2p/simulations/http.go b/p2p/simulations/http.go index 9ae935e08d..cda7493dc5 100644 --- a/p2p/simulations/http.go +++ b/p2p/simulations/http.go @@ -59,6 +59,17 @@ func (c *Client) GetNetwork(networkID string) (*Network, error) { return network, c.Get(fmt.Sprintf("/networks/%s", networkID), network) } +// CreateSnapshot creates a network snapshot +func (c *Client) CreateSnapshot(networkID string) (*Snapshot, error) { + snap := &Snapshot{} + return snap, c.Post(fmt.Sprintf("/networks/%s/snapshots/create", networkID), nil, snap) +} + +// LoadSnapshot loads a snapshot into a network +func (c *Client) LoadSnapshot(networkID string, snap *Snapshot) error { + return c.Post(fmt.Sprintf("/networks/%s/snapshots/load", networkID), snap, nil) +} + // SubscribeNetwork subscribes to network events which are sent from the server // as a server-sent-events stream func (c *Client) SubscribeNetwork(networkID string, events chan *Event) (event.Subscription, error) { @@ -225,8 +236,9 @@ func (c *Client) Send(method, path string, in, out interface{}) error { // ServerConfig is the configuration used to start an API server type ServerConfig struct { - // Adapter is the NodeAdapter to use when creating new networks - Adapter adapters.NodeAdapter + // NewAdapter is called to create a new NodeAdapter for each new + // network + NewAdapter func() adapters.NodeAdapter // Mocker is the function which will be called when a client sends a // POST request to /networks//mock and is expected to @@ -246,8 +258,8 @@ type Server struct { // NewServer returns a new simulation API server func NewServer(config *ServerConfig) *Server { - if config.Adapter == nil { - panic("Adapter not set") + if config.NewAdapter == nil { + panic("NewAdapter not set") } s := &Server{ @@ -260,6 +272,8 @@ func NewServer(config *ServerConfig) *Server { s.GET("/networks", s.GetNetworks) s.GET("/networks/:netid", s.GetNetwork) s.GET("/networks/:netid/events", s.StreamNetworkEvents) + s.POST("/networks/:netid/snapshots/create", s.CreateSnapshot) + s.POST("/networks/:netid/snapshots/load", s.LoadSnapshot) s.POST("/networks/:netid/mock", s.StartMocker) s.POST("/networks/:netid/nodes", s.CreateNode) s.GET("/networks/:netid/nodes", s.GetNodes) @@ -290,7 +304,7 @@ func (s *Server) CreateNetwork(w http.ResponseWriter, req *http.Request) { if _, exists := s.networks[config.Id]; exists { return nil, fmt.Errorf("network exists: %s", config.Id) } - network := NewNetwork(s.Adapter, config) + network := NewNetwork(s.NewAdapter(), config) s.networks[config.Id] = network return network, nil }() @@ -381,6 +395,37 @@ func (s *Server) StreamNetworkEvents(w http.ResponseWriter, req *http.Request) { } } +// CreateSnapshot creates a network snapshot +func (s *Server) CreateSnapshot(w http.ResponseWriter, req *http.Request) { + network := req.Context().Value("network").(*Network) + + snap, err := network.Snapshot() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + s.JSON(w, http.StatusOK, snap) +} + +// LoadSnapshot loads a snapshot into a network +func (s *Server) LoadSnapshot(w http.ResponseWriter, req *http.Request) { + network := req.Context().Value("network").(*Network) + + snap := &Snapshot{} + if err := json.NewDecoder(req.Body).Decode(snap); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + if err := network.Load(snap); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + s.JSON(w, http.StatusOK, network) +} + // CreateNode creates a node in a network using the given configuration func (s *Server) CreateNode(w http.ResponseWriter, req *http.Request) { network := req.Context().Value("network").(*Network) diff --git a/p2p/simulations/http_test.go b/p2p/simulations/http_test.go index 0f1b603add..f2c31d48e7 100644 --- a/p2p/simulations/http_test.go +++ b/p2p/simulations/http_test.go @@ -2,6 +2,8 @@ package simulations import ( "context" + "fmt" + "math/rand" "net/http/httptest" "sync/atomic" "testing" @@ -16,10 +18,15 @@ import ( type testService struct { id *adapters.NodeId + + // state stores []byte used to test creating and loading snapshots + state atomic.Value } -func newTestService(id *adapters.NodeId) node.Service { - return &testService{id} +func newTestService(id *adapters.NodeId, snapshot []byte) node.Service { + svc := &testService{id: id} + svc.state.Store(snapshot) + return svc } func (t *testService) Protocols() []p2p.Protocol { @@ -35,7 +42,7 @@ func (t *testService) APIs() []rpc.API { return []rpc.API{{ Namespace: "test", Version: "1.0", - Service: &TestAPI{}, + Service: &TestAPI{state: &t.state}, }} } @@ -56,9 +63,14 @@ func (t *testService) Run(_ *p2p.Peer, rw p2p.MsgReadWriter) error { } } +func (t *testService) Snapshot() ([]byte, error) { + return t.state.Load().([]byte), nil +} + // TestAPI provides a simple API to get and increment a counter and to // subscribe to increment events type TestAPI struct { + state *atomic.Value counter int64 feed event.Feed } @@ -72,6 +84,14 @@ func (t *TestAPI) Add(delta int64) { t.feed.Send(delta) } +func (t *TestAPI) GetState() []byte { + return t.state.Load().([]byte) +} + +func (t *TestAPI) SetState(state []byte) { + t.state.Store(state) +} + func (t *TestAPI) Events(ctx context.Context) (*rpc.Subscription, error) { notifier, supported := rpc.NotifierFromContext(ctx) if !supported { @@ -106,14 +126,17 @@ var testServices = adapters.Services{ "test": newTestService, } +func testHTTPServer(t *testing.T) *httptest.Server { + return httptest.NewServer(NewServer(&ServerConfig{ + NewAdapter: func() adapters.NodeAdapter { return adapters.NewSimAdapter(testServices) }, + })) +} + // TestHTTPNetwork tests creating and interacting with a simulation // network using the HTTP API func TestHTTPNetwork(t *testing.T) { // start the server - srv := NewServer(&ServerConfig{ - Adapter: adapters.NewSimAdapter(testServices), - }) - s := httptest.NewServer(srv) + s := testHTTPServer(t) defer s.Close() // create a network @@ -203,42 +226,55 @@ func TestHTTPNetwork(t *testing.T) { } // check we got all the events - nodeEvent := func(id string, up bool) *Event { - return &Event{ - Type: EventTypeNode, - Node: &Node{ - Config: &adapters.NodeConfig{ - Id: adapters.NewNodeIdFromHex(id), - }, - Up: up, + x := &expectEvents{t, events, sub} + x.expect( + x.nodeEvent(nodeIDs[0], false), + x.nodeEvent(nodeIDs[1], false), + x.nodeEvent(nodeIDs[0], true), + x.nodeEvent(nodeIDs[1], true), + x.connEvent(nodeIDs[0], nodeIDs[1], false), + x.connEvent(nodeIDs[0], nodeIDs[1], true), + ) +} + +type expectEvents struct { + *testing.T + + events chan *Event + sub event.Subscription +} + +func (t *expectEvents) nodeEvent(id string, up bool) *Event { + return &Event{ + Type: EventTypeNode, + Node: &Node{ + Config: &adapters.NodeConfig{ + Id: adapters.NewNodeIdFromHex(id), }, - } + Up: up, + }, } - connEvent := func(one, other string, up bool) *Event { - return &Event{ - Type: EventTypeConn, - Conn: &Conn{ - One: adapters.NewNodeIdFromHex(one), - Other: adapters.NewNodeIdFromHex(other), - Up: up, - }, - } - } - expectedEvents := []*Event{ - nodeEvent(nodeIDs[0], false), - nodeEvent(nodeIDs[1], false), - nodeEvent(nodeIDs[0], true), - nodeEvent(nodeIDs[1], true), - connEvent(nodeIDs[0], nodeIDs[1], false), - connEvent(nodeIDs[0], nodeIDs[1], true), +} + +func (t *expectEvents) connEvent(one, other string, up bool) *Event { + return &Event{ + Type: EventTypeConn, + Conn: &Conn{ + One: adapters.NewNodeIdFromHex(one), + Other: adapters.NewNodeIdFromHex(other), + Up: up, + }, } +} + +func (t *expectEvents) expect(events ...*Event) { timeout := time.After(10 * time.Second) - for i := 0; i < len(expectedEvents); i++ { + for i := 0; i < len(events); i++ { select { - case event := <-events: + case event := <-t.events: t.Logf("received %s event: %s", event.Type, event) - expected := expectedEvents[i] + expected := events[i] if event.Type != expected.Type { t.Fatalf("expected event %d to have type %q, got %q", i, expected.Type, event.Type) } @@ -260,6 +296,12 @@ func TestHTTPNetwork(t *testing.T) { if event.Conn == nil { t.Fatal("expected event.Conn to be set") } + if event.Conn.One == nil { + t.Fatal("expected event.Conn.One to be set") + } + if event.Conn.Other == nil { + t.Fatal("expected event.Conn.Other to be set") + } if event.Conn.One.NodeID != expected.Conn.One.NodeID { t.Fatalf("expected conn event %d to have one=%q, got one=%q", i, expected.Conn.One.Label(), event.Conn.One.Label()) } @@ -271,7 +313,7 @@ func TestHTTPNetwork(t *testing.T) { } } - case err := <-sub.Err(): + case err := <-t.sub.Err(): t.Fatalf("network stream closed unexpectedly: %s", err) case <-timeout: @@ -283,10 +325,7 @@ func TestHTTPNetwork(t *testing.T) { // TestHTTPNodeRPC tests calling RPC methods on nodes via the HTTP API func TestHTTPNodeRPC(t *testing.T) { // start the server - srv := NewServer(&ServerConfig{ - Adapter: adapters.NewSimAdapter(testServices), - }) - s := httptest.NewServer(srv) + s := testHTTPServer(t) defer s.Close() // start a node in a network @@ -345,3 +384,129 @@ func TestHTTPNodeRPC(t *testing.T) { t.Fatal(ctx.Err()) } } + +// TestHTTPSnapshot tests creating and loading network snapshots +func TestHTTPSnapshot(t *testing.T) { + // start the server + s := testHTTPServer(t) + defer s.Close() + + // create a two-node network + client := NewClient(s.URL) + network, err := client.CreateNetwork(&NetworkConfig{DefaultService: "test"}) + if err != nil { + t.Fatalf("error creating network: %s", err) + } + nodeCount := 2 + nodes := make([]*p2p.NodeInfo, nodeCount) + for i := 0; i < nodeCount; i++ { + node, err := client.CreateNode(network.Id, &adapters.NodeConfig{}) + if err != nil { + t.Fatalf("error creating node: %s", err) + } + if err := client.StartNode(network.Id, node.ID); err != nil { + t.Fatalf("error starting node: %s", err) + } + nodes[i] = node + } + if err := client.ConnectNode(network.Id, nodes[0].ID, nodes[1].ID); err != nil { + t.Fatalf("error connecting nodes: %s", err) + } + + // store some state in the test services + states := make([]string, nodeCount) + for i, node := range nodes { + rpc, err := client.RPCClient(context.Background(), network.Id, node.ID) + if err != nil { + t.Fatalf("error getting RPC client: %s", err) + } + defer rpc.Close() + state := fmt.Sprintf("%x", rand.Int()) + if err := rpc.Call(nil, "test_setState", []byte(state)); err != nil { + t.Fatalf("error setting service state: %s", err) + } + states[i] = state + } + + // create a snapshot + snap, err := client.CreateSnapshot(network.Id) + if err != nil { + t.Fatalf("error creating snapshot: %s", err) + } + for i, state := range states { + if string(snap.Nodes[i].Snapshot) != state { + t.Fatalf("expected snapshot state %q, got %q", state, snap.Nodes[i].Snapshot) + } + } + + // create another network + network, err = client.CreateNetwork(&NetworkConfig{DefaultService: "test"}) + if err != nil { + t.Fatalf("error creating network: %s", err) + } + + // subscribe to events so we can check them later + events := make(chan *Event, 100) + sub, err := client.SubscribeNetwork(network.Id, events) + if err != nil { + t.Fatalf("error subscribing to network events: %s", err) + } + defer sub.Unsubscribe() + + // load the snapshot + if err := client.LoadSnapshot(network.Id, snap); err != nil { + t.Fatalf("error loading snapshot: %s", err) + } + + // check the nodes and connection exists + net, err := client.GetNetwork(network.Id) + if err != nil { + t.Fatalf("error getting network: %s", err) + } + if len(net.Nodes) != nodeCount { + t.Fatalf("expected network to have %d nodes, got %d", nodeCount, len(net.Nodes)) + } + for i, node := range nodes { + id := net.Nodes[i].ID().String() + if id != node.ID { + t.Fatalf("expected node %d to have ID %s, got %s", i, node.ID, id) + } + } + if len(net.Conns) != 1 { + t.Fatalf("expected network to have 1 connection, got %d", len(net.Conns)) + } + conn := net.Conns[0] + if conn.One.String() != nodes[0].ID { + t.Fatalf("expected connection to have one=%q, got one=%q", nodes[0].ID, conn.One) + } + if conn.Other.String() != nodes[1].ID { + t.Fatalf("expected connection to have other=%q, got other=%q", nodes[1].ID, conn.Other) + } + + // check the node states were restored + for i, node := range nodes { + rpc, err := client.RPCClient(context.Background(), network.Id, node.ID) + if err != nil { + t.Fatalf("error getting RPC client: %s", err) + } + defer rpc.Close() + var state []byte + if err := rpc.Call(&state, "test_getState"); err != nil { + t.Fatalf("error getting service state: %s", err) + } + if string(state) != states[i] { + t.Fatalf("expected snapshot state %q, got %q", states[i], state) + } + } + + // check we got all the events + x := &expectEvents{t, events, sub} + x.expect( + x.nodeEvent(nodes[0].ID, false), + x.nodeEvent(nodes[0].ID, true), + x.nodeEvent(nodes[1].ID, false), + x.nodeEvent(nodes[1].ID, true), + x.connEvent(nodes[0].ID, nodes[1].ID, false), + x.connEvent(nodes[0].ID, nodes[1].ID, true), + ) +} diff --git a/p2p/simulations/journal.go b/p2p/simulations/journal.go deleted file mode 100644 index a827d3957e..0000000000 --- a/p2p/simulations/journal.go +++ /dev/null @@ -1,235 +0,0 @@ -package simulations - -import ( - "bytes" - "encoding/json" - "fmt" - "sync" - "time" - - "github.com/ethereum/go-ethereum/event" - "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/p2p/simulations/adapters" -) - -// Journal is an instance of a guaranteed no-loss subscription to network related events -// (using event.TypeMux). Network components POST events to the TypeMux, which then is -// read by the journal. Each journal belongs to a subscription. -type Journal struct { - Id string `json:"id"` - Events []*Event `json:"events"` - - lock sync.Mutex - counter int - cursor int - quitc chan bool -} - -// NewJournal constructor -// Journal can get input events from subscriptions, add event logs -// or scheduled replay of events from another journal -// -// see the Read and TimedRead iterators for use -// the Journal is safe for concurrent reads and writes -func NewJournal() *Journal { - return &Journal{quitc: make(chan bool)} -} - -// Subscribe subscribes to an event.Feed and launches a gorourine that appends -// any new event to the event log used for journalling history of a network -// the goroutine terminates when the journal is closed -func (self *Journal) Subscribe(feed *event.Feed) { - log.Info("subscribe") - events := make(chan *Event) - sub := feed.Subscribe(events) - go func() { - defer sub.Unsubscribe() - for { - select { - case event := <-events: - self.append(event) - case <-self.quitc: - return - } - } - }() -} - -// AddJournal appends the event log of another journal to the receiver's one -func (self *Journal) AddJournal(j *Journal) { - self.append(j.Events...) -} - -// NewJournalFromJSON decodes a JSON serialised events log -// into a journal struct -// used to replay recorded history -func NewJournalFromJSON(b []byte) (*Journal, error) { - self := NewJournal() - err := json.Unmarshal(b, self) - if err != nil { - return nil, err - } - return self, nil -} - -// Replay replays the events of another journal preserving (relative) timing of events -// params: -// * acc: using acceleration factor acc -// * journal: journal to use -// * feed: where to post the replayed events -func Replay(acc float64, j *Journal, feed *event.Feed) { - f := func(event *Event) bool { - // reposts the data with the eventer (the data receives a new timestamp) - event.Time = time.Now() - feed.Send(event) - return true - } - j.TimedRead(acc, f) -} - -// Snapshot creates a snapshot out of the journal -// this is simply done by reading the event log backwards and mark the last action -// on a node/connection ignoring all earlier mentions -// TODO: implmented -func Snapshot(conf *SnapshotConfig, j *Journal) (*Journal, error) { - return nil, fmt.Errorf("snapshot not implemented") -} - -func (self *Journal) Close() { - close(self.quitc) -} - -func (self *Journal) append(events ...*Event) { - self.lock.Lock() - defer self.lock.Unlock() - self.Events = append(self.Events, events...) - self.counter++ -} - -func (self *Journal) NewEntries() int { - self.lock.Lock() - defer self.lock.Unlock() - return self.counter - self.cursor -} - -func (self *Journal) WaitEntries(n int) { - for self.NewEntries() < n { - time.Sleep(10 * time.Millisecond) - } -} - -func (self *Journal) Read(f func(*Event) bool) (read int) { - self.lock.Lock() - defer self.lock.Unlock() - ok := true - for self.cursor < len(self.Events) && ok { - read++ - ok = f(self.Events[self.cursor]) - self.cursor++ - select { - case <-self.quitc: - break - default: - } - } - self.reset(self.cursor) - return read -} - -// TimedRead reads the events but blocks for intervals that correspond to -// the original time intervals, -// NOTE: the events' timestamps are supposed to be strictly ordered otherwise -// the call panics. -// acc is an acceleration factor -func (self *Journal) TimedRead(acc float64, f func(*Event) bool) (read int) { - var lastEvent time.Time - timer := time.NewTimer(0) - var event *Event - h := func(e *Event) bool { - // wait for the interval time passes event time - if e.Time.Before(lastEvent) { - panic("events not ordered") - } - interval := e.Time.Sub(lastEvent) - log.Trace(fmt.Sprintf("reset timer to interval %v", interval)) - timer.Reset(time.Duration(acc) * interval) - lastEvent = e.Time - event = e - return false - } - var n int - for { - // Read blocks for the iteration. need to read one event at a time so that - // waiting for the timer to go off does not block concurrent access to the journal - n = self.Read(h) - if read > 0 && n > 0 { - select { - case <-self.quitc: - break - case <-timer.C: - } - } - read += n - if n == 0 || !f(event) { - log.Trace(fmt.Sprintf("timed read ends (read %v entries)", read)) - break - } - } - return read -} - -func (self *Journal) Reset(n int) { - self.lock.Lock() - defer self.lock.Unlock() - self.reset(n) -} - -func (self *Journal) reset(n int) { - length := len(self.Events) - if length == 0 { - return - } - if n >= length-1 { - n = length - 1 - } - log.Trace(fmt.Sprintf("cursor reset from %v to %v/%v (%v)", self.cursor, n, len(self.Events), self.counter)) - self.Events = self.Events[self.cursor:] - self.cursor = 0 -} - -func (self *Journal) Counter() int { - self.lock.Lock() - defer self.lock.Unlock() - return self.counter -} - -// type History() - -func (self *Journal) Cursor() int { - self.lock.Lock() - defer self.lock.Unlock() - return self.cursor -} - -type SnapshotConfig struct { - Id string -} - -type JournalPlayConfig struct { - Id string - SpeedUp float64 - Journal *Journal - Events []string -} - -func ConnLabel(source, target *adapters.NodeId) string { - var first, second *adapters.NodeId - if bytes.Compare(source.Bytes(), target.Bytes()) > 0 { - first = target - second = source - } else { - first = source - second = target - } - return fmt.Sprintf("%v-%v", first, second) -} diff --git a/p2p/simulations/journal_test.go b/p2p/simulations/journal_test.go deleted file mode 100644 index 1c25ea7900..0000000000 --- a/p2p/simulations/journal_test.go +++ /dev/null @@ -1,143 +0,0 @@ -package simulations - -import ( - "encoding/json" - "io/ioutil" - "testing" - "time" - - "github.com/ethereum/go-ethereum/event" - "github.com/ethereum/go-ethereum/p2p/simulations/adapters" -) - -func testEvents(intervals ...int) (events []*Event) { - t := time.Now() - for _, interval := range intervals { - t = t.Add(time.Duration(interval) * time.Millisecond) - events = append(events, &Event{ - Type: EventTypeNode, - Time: t, - Node: &Node{Up: false}, - }) - } - return events -} - -func TestTimedRead(t *testing.T) { - j := NewJournal() - intervals := []int{100, 200, 300, 300, 100, 200} - j.Events = testEvents(intervals...) - var newTimes []time.Time - var i int - acc := 0.5 - length := 4 - f := func(event *Event) bool { - newTimes = append(newTimes, time.Now()) - i++ - return i <= length - } - start := time.Now() - read := j.TimedRead(acc, f) - if read != 5 { - t.Fatalf("incorrect number of events read: expected 5, got %v", read) - } - for i, ti := range newTimes { - expInt := time.Duration(acc*float64(intervals[i])) * time.Millisecond - gotInt := ti.Sub(start) - if gotInt-expInt > 1*time.Millisecond { - t.Fatalf("journal timed read incorrect interval: expected %v ,got %v", expInt, gotInt) - } - start = ti - } -} - -func testIDs() (ids []*adapters.NodeId) { - - keys := []string{ - "aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80", - "f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3", - } - for _, key := range keys { - id := adapters.NewNodeIdFromHex(key) - ids = append(ids, id) - } - return ids -} - -func testJournal(ids []*adapters.NodeId) *Journal { - eventer := &event.Feed{} - journal := NewJournal() - journal.Subscribe(eventer) - for _, id := range ids { - eventer.Send(&Event{ - Type: EventTypeNode, - Node: &Node{Config: &adapters.NodeConfig{Id: id}, Up: true}, - }) - } - journal.WaitEntries(len(ids)) - return journal -} - -func TestSubscribe(t *testing.T) { - ids := testIDs() - journal := testJournal(ids) - for i, ev := range journal.Events { - id := ev.Node.ID() - if id != ids[i] { - t.Fatalf("incorrect id: expected %v, got %v", id, ids[i]) - } - } -} - -func loadTestJournal(t *testing.T) ([]byte, *Journal) { - b, err := ioutil.ReadFile("./testjournal.json") - if err != nil { - t.Fatalf("unexpected error reading test journal json: %v", err) - } - journal, err := NewJournalFromJSON(b) - if err != nil { - t.Fatalf("unexpected error decoding journal json: %v", err) - } - return b, journal -} - -func TestLoadSave(t *testing.T) { - b, j := loadTestJournal(t) - - jo, err := json.MarshalIndent(j, "", " ") - if err != nil { - t.Fatalf("unexpected error encoding journal for %v: %v", j, err) - } - expJSON := string(b) - gotJSON := string(jo) - if expJSON != gotJSON { - t.Fatalf("incorrect json for journal: expected %v, got %v", expJSON, gotJSON) - } -} - -func TestReplay(t *testing.T) { - _, jo := loadTestJournal(t) - eventer := &event.Feed{} - - journal := NewJournal() - journal.Subscribe(eventer) - - Replay(0, jo, eventer) - for i, ev := range jo.Events { - exp := ev.String() - got := journal.Events[i].String() - if exp != got { - t.Fatalf("incorrent replayed journal entry at pos %v: expected %v, got %v", i, exp, got) - } - } - // ids := RandomNodeIds(7) - // ticker := time.NewTicker(1000 * time.Microsecond) - // go MockEvents(eventer, ids, ticker.C) - // journal.WaitEntries(20) - // // eventer.Stop() // eventer = &event.TypeMux{} - // journal = NewJournal() - // journal.Subscribe(eventer, &Entry{}) - - // func TestReplay(t *testing.T) { - // } -} diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index bed5ae5cb0..0adc696ca8 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -27,6 +27,7 @@ package simulations import ( + "bytes" "context" "fmt" "sync" @@ -261,6 +262,10 @@ func (self *Conn) nodesUp() 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) +} + +func (self *Network) startWithSnapshot(id *adapters.NodeId, snapshot []byte) error { node := self.GetNode(id) if node == nil { return fmt.Errorf("node %v does not exist", id) @@ -269,7 +274,7 @@ func (self *Network) Start(id *adapters.NodeId) error { 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(); err != nil { + if err := node.Start(snapshot); err != nil { return err } node.Up = true @@ -571,3 +576,75 @@ func (self *Network) Shutdown() { } } } + +func ConnLabel(source, target *adapters.NodeId) string { + var first, second *adapters.NodeId + if bytes.Compare(source.Bytes(), target.Bytes()) > 0 { + first = target + second = source + } else { + first = source + second = target + } + return fmt.Sprintf("%v-%v", first, second) +} + +// Snapshot represents the state of a network at a single point in time and can +// be used to restore the state of a network +type Snapshot struct { + Nodes []NodeSnapshot `json:"nodes,omitempty"` + Conns []Conn `json:"conns,omitempty"` +} + +// NodeSnapshot represents the state of a node in the network +type NodeSnapshot struct { + Node + + // Snapshot is arbitrary data gathered from calling node.Snapshot() + Snapshot []byte `json:"snapshot,omitempty"` +} + +// Snapshot creates a network snapshot +func (self *Network) Snapshot() (*Snapshot, error) { + self.lock.Lock() + defer self.lock.Unlock() + snap := &Snapshot{ + Nodes: make([]NodeSnapshot, len(self.Nodes)), + Conns: make([]Conn, len(self.Conns)), + } + for i, node := range self.Nodes { + snap.Nodes[i] = NodeSnapshot{Node: *node} + if !node.Up { + continue + } + snapshot, err := node.Snapshot() + if err != nil { + return nil, err + } + snap.Nodes[i].Snapshot = snapshot + } + for i, conn := range self.Conns { + snap.Conns[i] = *conn + } + return snap, nil +} + +func (self *Network) Load(snap *Snapshot) error { + for _, node := range snap.Nodes { + if _, err := self.NewNodeWithConfig(node.Config); err != nil { + return err + } + if !node.Up { + continue + } + if err := self.startWithSnapshot(node.Config.Id, node.Snapshot); err != nil { + return err + } + } + for _, conn := range snap.Conns { + if err := self.Connect(conn.One, conn.Other); err != nil { + return err + } + } + return nil +} diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index 51364dfb15..22d97efec0 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -22,7 +22,7 @@ import ( const serviceName = "discovery" var services = adapters.Services{ - serviceName: func(id *adapters.NodeId) p2pnode.Service { + serviceName: func(id *adapters.NodeId, snapshot []byte) p2pnode.Service { return newNode(id) }, } diff --git a/swarm/network/simulations/overlay.go b/swarm/network/simulations/overlay.go index 69b42cbd92..5dcfea3574 100644 --- a/swarm/network/simulations/overlay.go +++ b/swarm/network/simulations/overlay.go @@ -55,7 +55,7 @@ func (self *SimNode) Stop() error { } // NewSimNode creates adapters for nodes in the simulation. -func NewSimNode(id *adapters.NodeId) node.Service { +func NewSimNode(id *adapters.NodeId, snapshot []byte) node.Service { addr := network.NewPeerAddrFromNodeId(id) kp := network.NewKadParams() @@ -155,8 +155,8 @@ func main() { adapters.RegisterServices(services) config := &simulations.ServerConfig{ - Adapter: adapters.NewSimAdapter(services), - Mocker: mocker, + NewAdapter: func() adapters.NodeAdapter { return adapters.NewSimAdapter(services) }, + Mocker: mocker, } log.Info("starting simulation server on 0.0.0.0:8888...")