From 4ffab91adf314f41fbc562b1ff93dad29d677f1b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kurk=C3=B3=20Mih=C3=A1ly?= Date: Thu, 21 Mar 2019 19:46:32 +0200 Subject: [PATCH] cmd, dashboard, p2p: patch peer event data race --- cmd/geth/config.go | 30 +++++++++++- cmd/utils/flags.go | 4 +- dashboard/dashboard.go | 9 +++- dashboard/peers.go | 106 +++++++++++++++++++++++++++-------------- p2p/metrics.go | 5 ++ 5 files changed, 112 insertions(+), 42 deletions(-) diff --git a/cmd/geth/config.go b/cmd/geth/config.go index f316380cee..53c19bf1c2 100644 --- a/cmd/geth/config.go +++ b/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) diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index e00f92fa75..f2233777a5 100644 --- a/cmd/utils/flags.go +++ b/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 }) } diff --git a/dashboard/dashboard.go b/dashboard/dashboard.go index d69a750f10..8f0fff6281 100644 --- a/dashboard/dashboard.go +++ b/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, } } diff --git a/dashboard/peers.go b/dashboard/peers.go index e591bda779..9b822c296b 100644 --- a/dashboard/peers.go +++ b/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)) diff --git a/p2p/metrics.go b/p2p/metrics.go index 2b1ad6df41..2a8b6b6921 100644 --- a/p2p/metrics.go +++ b/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 {