cmd, dashboard, p2p: patch peer event data race

cmd/geth/config.go

@@ -25,6 +25,9 @@ import (
 	"reflect"
 	"unicode"
 
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p"
+
 	cli "gopkg.in/urfave/cli.v1"
 
 	"github.com/ethereum/go-ethereum/cmd/utils"
@@ -160,7 +163,32 @@ func makeFullNode(ctx *cli.Context) *node.Node {
 	utils.RegisterEthService(stack, &cfg.Eth)
 
 	if ctx.GlobalBool(utils.DashboardEnabledFlag.Name) {
-		utils.RegisterDashboardService(stack, &cfg.Dashboard, gitCommit)
+		// There is a data race between the network layer and the dashboard, which
+		// can cause some lost peer events, therefore some peers might not appear
+		// on the dashboard.
+		// In order to solve this problem, a peer event subscription is registered
+		// before the network layer starts, and when the dashboard is ready, the
+		// stored events are passed to it.
+		peerEventBridge := make(chan p2p.MeteredPeerEvent, 1000) // The events are stored by and passed through this channel.
+		closePeerEventBridge := make(chan struct{})              // This channel gets a signal from the dashboard when it is ready.
+		go func() {
+			peerCh := make(chan p2p.MeteredPeerEvent, 200)   // Channel for the initial peer events.
+			subPeer := p2p.SubscribeMeteredPeerEvent(peerCh) // Subscribe to the peer events.
+			for {
+				select {
+				case event := <-peerCh:
+					select {
+					case peerEventBridge <- event:
+					default:
+						log.Warn("Too many peer events before the dashboard starts")
+					}
+				case <-closePeerEventBridge:
+					subPeer.Unsubscribe()
+					return
+				}
+			}
+		}()
+		utils.RegisterDashboardService(stack, &cfg.Dashboard, gitCommit, peerEventBridge, closePeerEventBridge)
 	}
 	// Whisper must be explicitly enabled by specifying at least 1 whisper flag or in dev mode
 	shhEnabled := enableWhisper(ctx)

cmd/utils/flags.go

@@ -1461,9 +1461,9 @@ func RegisterEthService(stack *node.Node, cfg *eth.Config) {
 }
 
 // RegisterDashboardService adds a dashboard to the stack.
-func RegisterDashboardService(stack *node.Node, cfg *dashboard.Config, commit string) {
+func RegisterDashboardService(stack *node.Node, cfg *dashboard.Config, commit string, peerEventBridge chan p2p.MeteredPeerEvent, closePeerEventBridge chan struct{}) {
 	stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
-		return dashboard.New(cfg, commit, ctx.ResolvePath("logs")), nil
+		return dashboard.New(cfg, commit, ctx.ResolvePath("logs"), peerEventBridge, closePeerEventBridge), nil
 	})
 }
 

dashboard/dashboard.go

@@ -67,6 +67,9 @@ type Dashboard struct {
 
 	quit chan chan error // Channel used for graceful exit
 	wg   sync.WaitGroup  // Wait group used to close the data collector threads
+
+	peerEventBridge      chan p2p.MeteredPeerEvent // Channel for the initial peer events.
+	closePeerEventBridge chan struct{}             // Channel to signal, that the initial peer event collection can be stopped.
 }
 
 // client represents active websocket connection with a remote browser.
@@ -77,7 +80,7 @@ type client struct {
 }
 
 // New creates a new dashboard instance with the given configuration.
-func New(config *Config, commit string, logdir string) *Dashboard {
+func New(config *Config, commit string, logdir string, peerEventBridge chan p2p.MeteredPeerEvent, closePeerEventBridge chan struct{}) *Dashboard {
 	now := time.Now()
 	versionMeta := ""
 	if len(params.VersionMeta) > 0 {
@@ -103,7 +106,9 @@ func New(config *Config, commit string, logdir string) *Dashboard {
 				DiskWrite:      emptyChartEntries(now, sampleLimit),
 			},
 		},
-		logdir: logdir,
+		logdir:               logdir,
+		peerEventBridge:      peerEventBridge,
+		closePeerEventBridge: closePeerEventBridge,
 	}
 }
 

dashboard/peers.go

@@ -380,10 +380,6 @@ func (db *Dashboard) collectPeerData() {
 	}
 	defer db.geodb.close()
 
-	peerCh := make(chan p2p.MeteredPeerEvent, eventBufferLimit) // Peer event channel.
-	subPeer := p2p.SubscribeMeteredPeerEvent(peerCh)            // Subscribe to peer events.
-	defer subPeer.Unsubscribe()                                 // Unsubscribe at the end.
-
 	ticker := time.NewTicker(db.config.Refresh)
 	defer ticker.Stop()
 
@@ -428,43 +424,79 @@ func (db *Dashboard) collectPeerData() {
 	ingress, egress := new(trafficMap), new(trafficMap)
 	*ingress, *egress = make(trafficMap), make(trafficMap)
 
+	// handlePeerEvent handles a metered peer event.
+	handlePeerEvent := func(event p2p.MeteredPeerEvent) {
+		now := time.Now()
+		switch event.Type {
+		case p2p.PeerConnected:
+			connected := now.Add(-event.Elapsed)
+			newPeerEvents = append(newPeerEvents, &peerEvent{
+				IP:        event.IP.String(),
+				ID:        event.ID.String(),
+				Connected: &connected,
+			})
+		case p2p.PeerDisconnected:
+			ip, id := event.IP.String(), event.ID.String()
+			newPeerEvents = append(newPeerEvents, &peerEvent{
+				IP:           ip,
+				ID:           id,
+				Disconnected: &now,
+			})
+			// The disconnect event comes with the last metered traffic count,
+			// because after the disconnection the peer's meter is removed
+			// from the registry. It can happen, that between two metering
+			// period the same peer disconnects multiple times, and appending
+			// all the samples to the traffic arrays would shift the metering,
+			// so only the last metering is stored, overwriting the previous one.
+			ingress.insert(ip, id, float64(event.Ingress))
+			egress.insert(ip, id, float64(event.Egress))
+		case p2p.PeerHandshakeFailed:
+			connected := now.Add(-event.Elapsed)
+			newPeerEvents = append(newPeerEvents, &peerEvent{
+				IP:           event.IP.String(),
+				Connected:    &connected,
+				Disconnected: &now,
+			})
+		default:
+			log.Error("Unknown metered peer event type", "type", event.Type)
+		}
+	}
+
+	peerCh := make(chan p2p.MeteredPeerEvent, eventBufferLimit) // Peer event channel.
+	subPeer := p2p.SubscribeMeteredPeerEvent(peerCh)            // Subscribe to peer events.
+	defer subPeer.Unsubscribe()                                 // Unsubscribe at the end.
+
+	firstEvent := true // denotes whether the delivered peer event is the first one for the dashboard.
 	for {
 		select {
 		case event := <-peerCh:
-			now := time.Now()
-			switch event.Type {
-			case p2p.PeerConnected:
-				connected := now.Add(-event.Elapsed)
-				newPeerEvents = append(newPeerEvents, &peerEvent{
-					IP:        event.IP.String(),
-					ID:        event.ID.String(),
-					Connected: &connected,
-				})
-			case p2p.PeerDisconnected:
-				ip, id := event.IP.String(), event.ID.String()
-				newPeerEvents = append(newPeerEvents, &peerEvent{
-					IP:           ip,
-					ID:           id,
-					Disconnected: &now,
-				})
-				// The disconnect event comes with the last metered traffic count,
-				// because after the disconnection the peer's meter is removed
-				// from the registry. It can happen, that between two metering
-				// period the same peer disconnects multiple times, and appending
-				// all the samples to the traffic arrays would shift the metering,
-				// so only the last metering is stored, overwriting the previous one.
-				ingress.insert(ip, id, float64(event.Ingress))
-				egress.insert(ip, id, float64(event.Egress))
-			case p2p.PeerHandshakeFailed:
-				connected := now.Add(-event.Elapsed)
-				newPeerEvents = append(newPeerEvents, &peerEvent{
-					IP:           event.IP.String(),
-					Connected:    &connected,
-					Disconnected: &now,
-				})
-			default:
-				log.Error("Unknown metered peer event type", "type", event.Type)
+			if firstEvent {
+				// There is a data race between the network layer and the dashboard, which
+				// can cause some lost peer events, therefore some peers might not appear
+				// on the dashboard.
+				// In order to solve this problem, a peer event subscription is registered
+				// before the network layer starts, and when the dashboard is ready, the
+				// stored events are passed to it.
+				//
+				// In order to synchronize the two subscriptions, the stored events are
+				// processed until the first event of the dashboard is found. After that
+				// all the events will be delivered to the dashboard too.
+			sync:
+				for {
+					select {
+					case e := <-db.peerEventBridge:
+						if event.Equal(e) {
+							db.closePeerEventBridge <- struct{}{}
+							break sync
+						}
+						handlePeerEvent(e)
+					default: // There were no events before the dashboard started.
+						break sync
+					}
+				}
+				firstEvent = false
 			}
+			handlePeerEvent(event)
 		case <-ticker.C:
 			// Collect the traffic samples from the registry.
 			p2p.PeerIngressRegistry.Each(collectIngress(ingress))

p2p/metrics.go

@@ -80,6 +80,11 @@ type MeteredPeerEvent struct {
 	Egress  uint64               // Egress count at the moment of the event
 }
 
+// Equal reports whether event and e are equal.
+func (event *MeteredPeerEvent) Equal(e MeteredPeerEvent) bool {
+	return event.Type == e.Type && event.IP.Equal(e.IP) && event.ID == e.ID && event.Elapsed == e.Elapsed && event.Ingress == e.Ingress && event.Egress == e.Egress
+}
+
 // SubscribeMeteredPeerEvent registers a subscription for peer life-cycle events
 // if metrics collection is enabled.
 func SubscribeMeteredPeerEvent(ch chan<- MeteredPeerEvent) event.Subscription {