swarm: worked in feedback comments to PR

This commit is contained in:
Fabio Barone 2018-02-06 11:24:29 -05:00
parent 9e1aeac485
commit 18de39d462
2 changed files with 9 additions and 28 deletions

View file

@ -28,7 +28,6 @@ import (
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
"time"
"github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
@ -381,13 +380,6 @@ func (s *Server) ResetNetwork(w http.ResponseWriter, req *http.Request) {
// StreamNetworkEvents streams network events as a server-sent-events stream // StreamNetworkEvents streams network events as a server-sent-events stream
func (s *Server) StreamNetworkEvents(w http.ResponseWriter, req *http.Request) { 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) events := make(chan *Event)
sub := s.network.events.Subscribe(events) sub := s.network.events.Subscribe(events)
defer sub.Unsubscribe() defer sub.Unsubscribe()
@ -463,21 +455,9 @@ 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 { for {
select { select {
case event := <-events: 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 // only send message events which match the filters
if event.Msg != nil && !filters.Match(event.Msg) { if event.Msg != nil && !filters.Match(event.Msg) {
continue continue

View file

@ -103,12 +103,13 @@ func startStop(net *Network, quit chan struct{}, nodeCount int) {
func probabilistic(net *Network, quit chan struct{}, nodeCount int) { func probabilistic(net *Network, quit chan struct{}, nodeCount int) {
nodes, err := connectNodesInRing(net, nodeCount) nodes, err := connectNodesInRing(net, nodeCount)
if err != nil { if err != nil {
log.Error(fmt.Sprintf("%s", err)) select {
if _, ok := <-quit; ok == false { case <-quit:
//error may be due to abortion of mocking; so the quit channel is closed //error may be due to abortion of mocking; so the quit channel is closed
return return
default:
panic("Could not startup node network for mocker")
} }
panic("Could not startup node network for mocker")
} }
for { for {
select { select {
@ -149,7 +150,7 @@ func probabilistic(net *Network, quit chan struct{}, nodeCount int) {
log.Debug(fmt.Sprintf("node %v shutting down", nodes[i])) log.Debug(fmt.Sprintf("node %v shutting down", nodes[i]))
err := net.Stop(nodes[i]) err := net.Stop(nodes[i])
if err != nil { if err != nil {
log.Error(fmt.Sprintf("Error stopping node %s", nodes[i])) log.Error("Error stopping node", "node", nodes[i])
wg.Done() wg.Done()
continue continue
} }
@ -157,7 +158,7 @@ func probabilistic(net *Network, quit chan struct{}, nodeCount int) {
time.Sleep(randWait) time.Sleep(randWait)
err := net.Start(id) err := net.Start(id)
if err != nil { if err != nil {
log.Error(fmt.Sprintf("Error starting node %s", id)) log.Error("Error starting node", "node", id)
} }
wg.Done() wg.Done()
}(nodes[i]) }(nodes[i])
@ -174,7 +175,7 @@ func connectNodesInRing(net *Network, nodeCount int) ([]discover.NodeID, error)
conf := adapters.RandomNodeConfig() conf := adapters.RandomNodeConfig()
node, err := net.NewNodeWithConfig(conf) node, err := net.NewNodeWithConfig(conf)
if err != nil { if err != nil {
log.Error(fmt.Sprintf("Error creating a node! %s", err)) log.Error("Error creating a node!", "err", err)
return nil, err return nil, err
} }
ids[i] = node.ID() ids[i] = node.ID()
@ -182,7 +183,7 @@ func connectNodesInRing(net *Network, nodeCount int) ([]discover.NodeID, error)
for _, id := range ids { for _, id := range ids {
if err := net.Start(id); err != nil { if err := net.Start(id); err != nil {
log.Error("Error starting a node! %s", err) log.Error("Error starting a node!", "err", err)
return nil, err return nil, err
} }
log.Debug(fmt.Sprintf("node %v starting up", id)) log.Debug(fmt.Sprintf("node %v starting up", id))
@ -190,7 +191,7 @@ func connectNodesInRing(net *Network, nodeCount int) ([]discover.NodeID, error)
for i, id := range ids { for i, id := range ids {
peerID := ids[(i+1)%len(ids)] peerID := ids[(i+1)%len(ids)]
if err := net.Connect(id, peerID); err != nil { if err := net.Connect(id, peerID); err != nil {
log.Error(fmt.Sprintf("Error connecting a node to a peer! %s", err)) log.Error("Error connecting a node to a peer!", "err", err)
return nil, err return nil, err
} }
} }