diff --git a/p2p/simulations/simulation.go b/p2p/simulations/simulation.go new file mode 100644 index 0000000000..41f955bc8c --- /dev/null +++ b/p2p/simulations/simulation.go @@ -0,0 +1,132 @@ +package simulations + +import ( + "context" + "time" + + "github.com/ethereum/go-ethereum/p2p/adapters" +) + +// Simulation provides a framework for running actions in a simulated network +// and then waiting for expectations to be met +type Simulation struct { + network *Network +} + +// NewSimulation returns a new simulation +func NewSimulation(network *Network) *Simulation { + return &Simulation{ + network: network, + } +} + +// Run performs a step of the simulation by performing the step's action and +// then waiting for the step's expectation to be met +func (s *Simulation) Run(ctx context.Context, step *Step) (result *StepResult) { + result = newStepResult() + + result.StartedAt = time.Now() + defer func() { result.FinishedAt = time.Now() }() + + // watch network events for the duration of the step + stop := s.watchNetwork(result) + defer stop() + + // perform the action + if err := step.Action(ctx); err != nil { + result.Error = err + return + } + + // wait for all node expectations to either pass, error or timeout + for len(result.Passes) < len(step.Expect.Nodes) { + select { + case id := <-step.Trigger: + // skip if the node has already passed + if _, ok := result.Passes[id]; ok { + continue + } + + // run the node expectation check + pass, err := step.Expect.Check(ctx, id) + if err != nil { + result.Error = err + return + } + if pass { + result.Passes[id] = time.Now() + } + case <-ctx.Done(): + result.Error = ctx.Err() + return + } + } + + return +} + +func (s *Simulation) watchNetwork(result *StepResult) func() { + stop := make(chan struct{}) + done := make(chan struct{}) + sub := s.network.Events().Subscribe(ConnectivityEvents...) + go func() { + defer close(done) + defer sub.Unsubscribe() + events := sub.Chan() + for { + select { + case event := <-events: + result.NetworkEvents = append(result.NetworkEvents, event) + case <-stop: + return + } + } + }() + return func() { + close(stop) + <-done + } +} + +type Step struct { + // Action is the action to perform for this step + Action func(context.Context) error + + // Trigger is a channel which receives node ids and triggers an + // expectation check for that node + Trigger chan *adapters.NodeId + + // Expect is the expectation to wait for when performing this step + Expect *Expectation +} + +type Expectation struct { + // Nodes is a list of nodes to check + Nodes []*adapters.NodeId + + // Check checks whether a given node meets the expectation + Check func(context.Context, *adapters.NodeId) (bool, error) +} + +func newStepResult() *StepResult { + return &StepResult{ + Passes: make(map[*adapters.NodeId]time.Time), + } +} + +type StepResult struct { + // Error is the error encountered whilst running the step + Error error + + // StartedAt is the time the step started + StartedAt time.Time + + // FinishedAt is the time the step finished + FinishedAt time.Time + + // Passes are the timestamps of the successful node expectations + Passes map[*adapters.NodeId]time.Time + + // NetworkEvents are the network events which occurred during the step + NetworkEvents []interface{} +} diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go new file mode 100644 index 0000000000..b850642d0f --- /dev/null +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -0,0 +1,186 @@ +package discovery_test + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/adapters" + "github.com/ethereum/go-ethereum/p2p/simulations" + "github.com/ethereum/go-ethereum/swarm/network" +) + +func TestDiscoverySimulation(t *testing.T) { + // create 10 node network + nodeCount := 10 + trigger := make(chan *adapters.NodeId) + net := simulations.NewNetwork(&simulations.NetworkConfig{ + Id: "0", + Backend: true, + }) + nodes := make(map[*adapters.NodeId]*node, nodeCount) + net.SetNaf(func(conf *simulations.NodeConfig) adapters.NodeAdapter { + node := newNode(conf.Id, net, trigger) + nodes[conf.Id] = node + return node + }) + ids := adapters.RandomNodeIds(nodeCount) + for _, id := range ids { + net.NewNode(&simulations.NodeConfig{Id: id}) + if err := net.Start(id); err != nil { + t.Fatalf("error starting node %s: %s", id.Label(), err) + } + } + + // run a simulation which connects the 10 nodes in a ring and waits + // for full peer discovery + action := func(ctx context.Context) error { + for i, id := range ids { + var peerId *adapters.NodeId + if i == 0 { + peerId = ids[len(ids)-1] + } else { + peerId = ids[i-1] + } + if err := net.Connect(id, peerId); err != nil { + return err + } + } + return nil + } + check := func(ctx context.Context, id *adapters.NodeId) (bool, error) { + select { + case <-ctx.Done(): + return false, ctx.Err() + default: + } + + node, ok := nodes[id] + if !ok { + return false, fmt.Errorf("unknown node: %s", id) + } + + // TODO: check list of peers + _ = node + + return true, nil + } + + timeout := 10 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{ + Action: action, + Trigger: trigger, + Expect: &simulations.Expectation{ + Nodes: ids, + Check: check, + }, + }) + if result.Error != nil { + t.Fatalf("simulation failed: %s", result.Error) + } + + t.Log("Simulation Passed:") + t.Logf("Duration: %s", result.FinishedAt.Sub(result.StartedAt)) + for _, id := range ids { + t.Logf("Node %s passed in %s", id.Label(), result.Passes[id].Sub(result.StartedAt)) + } + t.Logf("Events:") + for _, event := range result.NetworkEvents { + t.Log(event) + } +} + +type node struct { + *network.Hive + adapters.NodeAdapter + + id *adapters.NodeId + network *simulations.Network + trigger chan *adapters.NodeId +} + +func newNode(id *adapters.NodeId, net *simulations.Network, trigger chan *adapters.NodeId) *node { + addr := network.NewPeerAddrFromNodeId(id) + kademlia := newKademlia(addr.OverlayAddr()) + hive := newHive(kademlia) + codeMap := network.BzzCodeMap(network.DiscoveryMsgs...) + nodeAdapter := adapters.NewSimNode(id, net, adapters.NewSimPipe) + node := &node{ + Hive: hive, + NodeAdapter: nodeAdapter, + id: id, + network: net, + trigger: trigger, + } + services := func(peer network.Peer) error { + discoveryPeer := network.NewDiscovery(peer, kademlia) + node.Add(discoveryPeer) + peer.DisconnectHook(func(err error) { + node.Remove(discoveryPeer) + }) + return nil + } + nodeAdapter.Run = network.Bzz(addr.OverlayAddr(), nodeAdapter, codeMap, services, nil, nil).Run + return node +} + +func newKademlia(overlayAddr []byte) *network.Kademlia { + params := network.NewKadParams() + params.MinProxBinSize = 2 + params.MaxBinSize = 3 + params.MinBinSize = 1 + params.MaxRetries = 1000 + params.RetryExponent = 2 + params.RetryInterval = 1000000 + + return network.NewKademlia(overlayAddr, params) +} + +func newHive(kademlia *network.Kademlia) *network.Hive { + params := network.NewHiveParams() + params.CallInterval = 5000 + + return network.NewHive(params, kademlia) +} + +func (n *node) Start() error { + return n.Hive.Start(n.connectPeer, n.hiveKeepAlive) +} + +func (n *node) Stop() error { + n.Hive.Stop() + return nil +} + +func (n *node) Add(peer network.Peer) error { + err := n.Hive.Add(peer) + n.triggerCheck() + return err +} + +func (n *node) Remove(peer network.Peer) { + n.Hive.Remove(peer) + n.triggerCheck() +} + +func (n *node) RunProtocol(id *adapters.NodeId, rw, rrw p2p.MsgReadWriter, peer *adapters.Peer) error { + return n.NodeAdapter.(adapters.ProtocolRunner).RunProtocol(id, rw, rrw, peer) +} + +func (n *node) connectPeer(s string) error { + return n.network.Connect(n.id, adapters.NewNodeIdFromHex(s)) +} + +func (n *node) hiveKeepAlive() <-chan time.Time { + return time.Tick(time.Second) +} + +func (n *node) triggerCheck() { + // TODO: rate limit the trigger? + go func() { n.trigger <- n.id }() +}