diff --git a/p2p/simulations/http.go b/p2p/simulations/http.go index 24001f1949..65332614a5 100644 --- a/p2p/simulations/http.go +++ b/p2p/simulations/http.go @@ -28,6 +28,7 @@ import ( "strconv" "strings" "sync" + "time" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/p2p" @@ -380,6 +381,13 @@ func (s *Server) ResetNetwork(w http.ResponseWriter, req *http.Request) { // StreamNetworkEvents streams network events as a server-sent-events stream func (s *Server) StreamNetworkEvents(w http.ResponseWriter, req *http.Request) { + var throttle <-chan time.Time + //When running a sim, the frontend may "choke to death" depending on params (node count) + //due to overwhelming event rate. It is therefore important to be able to throttle + //the rate at which the events are emitted, in order for the frontend to have time + //to terminate rendering before the next event arrives. + throttling := false + events := make(chan *Event) sub := s.network.events.Subscribe(events) defer sub.Unsubscribe() @@ -455,9 +463,21 @@ func (s *Server) StreamNetworkEvents(w http.ResponseWriter, req *http.Request) { } } + //Only do throttling if requested + if req.URL.Query().Get("throttling") == "true" { + throttling = true + //rate at which we allow events to be emitted on the server-side event stream + rate := time.Second / 40 + throttle = time.Tick(rate) + } + for { select { case event := <-events: + //if throttling, wait for the next throttle tick before emitting event + if throttling { + <-throttle + } // only send message events which match the filters if event.Msg != nil && !filters.Match(event.Msg) { continue diff --git a/p2p/simulations/mocker.go b/p2p/simulations/mocker.go index b370fe2cd2..349306ca3a 100644 --- a/p2p/simulations/mocker.go +++ b/p2p/simulations/mocker.go @@ -103,6 +103,11 @@ func startStop(net *Network, quit chan struct{}, nodeCount int) { func probabilistic(net *Network, quit chan struct{}, nodeCount int) { nodes, err := connectNodesInRing(net, nodeCount) if err != nil { + log.Error(fmt.Sprintf("%s", err)) + if _, ok := <-quit; ok == false { + //error may be due to abortion of mocking; so the quit channel is closed + return + } panic("Could not startup node network for mocker") } for { @@ -169,7 +174,7 @@ func connectNodesInRing(net *Network, nodeCount int) ([]discover.NodeID, error) conf := adapters.RandomNodeConfig() node, err := net.NewNodeWithConfig(conf) if err != nil { - log.Error("Error creating a node! %s", err) + log.Error(fmt.Sprintf("Error creating a node! %s", err)) return nil, err } ids[i] = node.ID() @@ -185,7 +190,7 @@ func connectNodesInRing(net *Network, nodeCount int) ([]discover.NodeID, error) for i, id := range ids { peerID := ids[(i+1)%len(ids)] if err := net.Connect(id, peerID); err != nil { - log.Error("Error connecting a node to a peer! %s", err) + log.Error(fmt.Sprintf("Error connecting a node to a peer! %s", err)) return nil, err } } diff --git a/swarm/network/simulations/overlay.go b/swarm/network/simulations/overlay.go index a82b1cd100..0c9ac4ae45 100644 --- a/swarm/network/simulations/overlay.go +++ b/swarm/network/simulations/overlay.go @@ -7,8 +7,6 @@ package main import ( "flag" - "fmt" - "math/rand" "net/http" "os" "runtime" @@ -72,162 +70,23 @@ func (s *Simulation) NewService(ctx *adapters.ServiceContext) (node.Service, err return network.NewBzz(config, kad, store, nil, nil), nil } -func createMockers() map[string]*simulations.MockerConfig { - configs := make(map[string]*simulations.MockerConfig) - - defaultCfg := simulations.DefaultMockerConfig() - defaultCfg.ID = "start-stop" - defaultCfg.Description = "Starts and Stops nodes in go routines" - defaultCfg.Mocker = startStopMocker - - bootNetworkCfg := simulations.DefaultMockerConfig() - bootNetworkCfg.ID = "bootNet" - bootNetworkCfg.Description = "Only boots up all nodes in the config" - bootNetworkCfg.Mocker = bootMocker - - randomNodesCfg := simulations.DefaultMockerConfig() - randomNodesCfg.ID = "randomNodes" - randomNodesCfg.Description = "Boots nodes and then starts and stops some picking randomly" - randomNodesCfg.Mocker = randomMocker - - configs[defaultCfg.ID] = defaultCfg - configs[bootNetworkCfg.ID] = bootNetworkCfg - configs[randomNodesCfg.ID] = randomNodesCfg - - return configs -} - -func setupMocker(net *simulations.Network) []discover.NodeID { - nodeCount := 30 - ids := make([]discover.NodeID, nodeCount) - for i := 0; i < nodeCount; i++ { - node, err := net.NewNode() - if err != nil { - panic(err.Error()) - } - ids[i] = node.ID() - } - - for _, id := range ids { - if err := net.Start(id); err != nil { - panic(err.Error()) - } - } - for i, id := range ids { - log.Trace(fmt.Sprintf("setup mocker: register a peer on node %x", id[:4])) - var peerID discover.NodeID - if i == 0 { - peerID = ids[len(ids)-1] - } else { - peerID = ids[i-1] - } - ch := make(chan network.OverlayAddr) - go func() { - defer close(ch) - ch <- network.NewAddrFromNodeID(peerID) - }() - log.Trace(fmt.Sprintf("%x registers peer %x", id[:4], peerID[:4])) - if err := net.GetNode(id).Node.(*adapters.SimNode).Services()[0].(*network.Bzz).Hive.Register(ch); err != nil { - panic(err.Error()) - } - } - - return ids -} - -func bootMocker(net *simulations.Network) { - setupMocker(net) -} - -func randomMocker(net *simulations.Network) { - ids := setupMocker(net) - - for { - var lowid, highid int - var wg sync.WaitGroup - randWait := rand.Intn(5000) + 1000 - rand1 := rand.Intn(9) - rand2 := rand.Intn(9) - if rand1 < rand2 { - lowid = rand1 - highid = rand2 - } else if rand1 > rand2 { - highid = rand1 - lowid = rand2 - } else { - if rand1 == 0 { - rand2 = 9 - } else if rand1 == 9 { - rand1 = 0 - } - lowid = rand1 - highid = rand2 - } - var steps = highid - lowid - wg.Add(steps) - for i := lowid; i < highid; i++ { - log.Info(fmt.Sprintf("node %v shutting down", ids[i])) - net.Stop(ids[i]) - go func(id discover.NodeID) { - time.Sleep(time.Duration(randWait) * time.Millisecond) - net.Start(id) - wg.Done() - }(ids[i]) - time.Sleep(time.Duration(randWait) * time.Millisecond) - } - wg.Wait() - } -} - -func startStopMocker(net *simulations.Network) { - ids := setupMocker(net) - - for range time.Tick(10 * time.Second) { - id := ids[rand.Intn(len(ids))] - go func() { - log.Error("stopping node", "id", id) - if err := net.Stop(id); err != nil { - log.Error("error stopping node", "id", id, "err", err) - return - } - - time.Sleep(3 * time.Second) - - log.Error("starting node", "id", id) - if err := net.Start(id); err != nil { - log.Error("error starting node", "id", id, "err", err) - return - } - }() - } -} - // var server func main() { flag.Parse() runtime.GOMAXPROCS(runtime.NumCPU()) - log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false)))) + log.Root().SetHandler(log.LvlFilterHandler(log.LvlInfo, log.StreamHandler(os.Stderr, log.TerminalFormat(false)))) s := NewSimulation() services := adapters.Services{ "overlay": s.NewService, } adapter := adapters.NewSimAdapter(services) - network := simulations.NewNetwork(adapter, &simulations.NetworkConfig{ DefaultService: "overlay", }) - mockers := createMockers() - - config := simulations.ServerConfig{ - DefaultMockerID: "randomNodes", - // DefaultMockerID: "bootNet", - Mockers: mockers, - } - log.Info("starting simulation server on 0.0.0.0:8888...") - http.ListenAndServe(":8888", simulations.NewServer(network, config)) + http.ListenAndServe(":8888", simulations.NewServer(network)) }