swarm: migrated overlay sim to rewrite

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

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
 		}
 	}

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))
 }