From 8c18fa4623a51bc8d84521785eb1f7413f1ce447 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kurk=C3=B3=20Mih=C3=A1ly?= Date: Thu, 20 Sep 2018 13:50:35 +0300 Subject: [PATCH] dashboard, p2p: experiment 3 --- dashboard/peers.go | 336 +++++++++++++++++++++------------------------ p2p/metrics.go | 88 ++++++------ 2 files changed, 203 insertions(+), 221 deletions(-) diff --git a/dashboard/peers.go b/dashboard/peers.go index 2a2cb60608..704b7dbd5a 100644 --- a/dashboard/peers.go +++ b/dashboard/peers.go @@ -17,7 +17,6 @@ package dashboard import ( - "container/list" "fmt" "time" @@ -25,60 +24,171 @@ import ( "github.com/ethereum/go-ethereum/p2p" ) -const eventBufferLimit = 128 // Maximum number of buffered peer events -const trafficEventBufferLimit = p2p.MeteredPeerLimit -const connectionLimit = 100 +const ( + eventBufferLimit = 128 // Maximum number of buffered peer events + connectionLimit = 16 +) var autoID int64 -type peerLimiter struct { - underlying *NetworkMessage - failed bool - l *list.List +// maintainedPeer is the element of the peer maintainer's linked list. +// Similarly to an ordinary linked list element it knows the previous +// and the next elements, and also has a pointer to its parent map in +// order to remove itself from there when it is removed from the list. +type maintainedPeer struct { + // To simplify the implementation, the list is implemented as a ring, + // such that 'root' is both the next element of the last list element + // and the previous element of the first list element. + prev *maintainedPeer // Pointer to the previous element + next *maintainedPeer // Pointer to the next element + parent *idContainer // Pointer to the parent + id string // NodeID of the peer } -func NewPeerLimiter(underlying *NetworkMessage, failed bool) *peerLimiter { - return &peerLimiter{l: list.New(), underlying: underlying, failed: failed} +// idContainer contains the NodeIDs belonging to an IP address. +// The pointer to the parent makes it possible for the container +// to remove itself from the parent's map when becomes empty. +type idContainer struct { + parent *PeerMaintainer // Pointer to the parent of the container + ip string // IP of the peer + peers map[string]*maintainedPeer // NodeIDs belonging to the given IP } -func (pl *peerLimiter) update(peer *Peer) { - return - if peer.element == nil { - peer.element = pl.l.PushBack(peer) +// update moves the list element belonging to the given ID to the end, +// or inserts a new element to the end of the list if the given ID didn't +// appear yet in the container. Returns the IP and the ID of the removed +// peer if an element removal happened, or nil otherwise. +func (idc *idContainer) update(id string) *removedPeer { + maintainer := idc.parent + if _, ok := idc.peers[id]; !ok { + e := &maintainedPeer{ + parent: idc, + id: id, + } + maintainer.insert(e, maintainer.root.prev) + idc.peers[id] = e } else { - pl.l.MoveToBack(peer.element) + maintainer.insert(maintainer.remove(idc.peers[id]), maintainer.root.prev) } - for pl.l.Len() > 2 {//p2p.MeteredPeerLimit { - pl.remove(pl.l.Front()) - } -} - -func (pl *peerLimiter) remove(e *list.Element) { - return - elem := pl.l.Remove(e) - if peer, ok := elem.(*Peer); ok { - if pl.failed { - fmt.Println(peer.ip, peer.id) - pl.underlying.PeerBundles[peer.ip].FailedPeers.remove(peer.id) - } else { - fmt.Println(peer.ip, peer.id[:10]) - pl.underlying.PeerBundles[peer.ip].Peers.remove(peer.id) + if maintainer.len > maintainer.limit { + first := maintainer.remove(maintainer.root.next) + return &removedPeer{ + ip: first.parent.ip, + id: first.id, } } + return nil } -func (pl *peerLimiter) clear() { - for pl.l.Front() != nil { - pl.remove(pl.l.Front()) +// remove removes the peer belonging to the given ID and removes itself too +// from the parent container if becomes empty. +func (idc *idContainer) remove(id string) { + delete(idc.peers, id) + if len(idc.peers) < 1 { + idc.parent.removeIP(idc.ip) + idc.parent = nil } } + +// PeerMaintainer is an abstract layer above the metered peer container, +// maintaining the peers. i.e. sorting them based on their activity and +// removing the oldest inactive ones when their count reaches the limit. // -//func tail(arr []time.Time) []time.Time { -// if first := len(arr)-connectionLimit; first > 0 { -// return arr[first:] -// } -// return arr -//} +// Consists of a map of maps which represent the peers grouped by the IP +// address then by the NodeID. The elements on the bottom of the tree are +// doubly linked and sorted by their activity. i.e. the first element is +// the peer that is inactive for the longest time. This makes it possible +// to count the peers and remove the oldest one effectively. +// +// When a peer event appears, the active peer goes to the end of the list +// and in case of removal the IP and the ID of the removed peer is returned. +type PeerMaintainer struct { + ids map[string]*idContainer // NodeIDs grouped by IP + root maintainedPeer // Sentinel list element + len int // Current list length excluding the sentinel element + limit int // Maximum number of maintained peers +} + +// init initializes the peer maintainer. +func (pm *PeerMaintainer) init(limit int) *PeerMaintainer { + if limit < 0 { + limit = 0 + } + pm.ids = make(map[string]*idContainer) + pm.root.prev = &pm.root + pm.root.next = &pm.root + pm.len = 0 + pm.limit = limit + return pm +} + +// NewPeerMaintainer returns an initialized peer maintainer. +func NewPeerMaintainer(limit int) *PeerMaintainer { + return new(PeerMaintainer).init(limit) +} + +// Update updates the peer element belonging to the given IP and ID. +func (pm *PeerMaintainer) Update(ip, id string) *removedPeer { + return pm.getOrInitIDs(ip).update(id) +} + +// getOrInitIDs returns the ID map. Initializes it if it doesn't exist. +func (pm *PeerMaintainer) getOrInitIDs(ip string) *idContainer { + if _, ok := pm.ids[ip]; !ok { + pm.ids[ip] = &idContainer{ + parent: pm, + ip: ip, + peers: make(map[string]*maintainedPeer), + } + } + return pm.ids[ip] +} + +// insert inserts 'e' after 'at' and increments the current list length. +func (pm *PeerMaintainer) insert(e, at *maintainedPeer) { + n := at.next + at.next = e + e.prev = at + e.next = n + n.prev = e + pm.len++ +} + +// remove removes e from the list and from the maintainer tree. +func (pm *PeerMaintainer) remove(e *maintainedPeer) *maintainedPeer { + e.next.prev = e.prev + e.prev.next = e.next + e.prev = nil + e.next = nil + e.parent.remove(e.id) + e.parent = nil + pm.len-- + return e +} + +// clear cleans up the peer maintainer. +func (pm *PeerMaintainer) clear() { + for pm.root.next != &pm.root { + next := pm.root.next + pm.root.next = next.next + next.prev = nil + next.next = nil + next.parent.remove(next.id) + next.parent = nil + } +} + +// removeIP removes the peers belonging to the given IP. It is supposed that +// 'ids' is empty, because the clearing direction is from bottom to top. +func (pm *PeerMaintainer) removeIP(ip string) { + delete(pm.ids, ip) +} + +// removedPeer contains the IP and the ID of the peer that was removed from +// the peer maintainer. +type removedPeer struct { + ip, id string +} // collectPeerData gathers data about the peers and sends it to the clients. func (db *Dashboard) collectPeerData() { @@ -93,154 +203,23 @@ func (db *Dashboard) collectPeerData() { } defer db.geodb.Close() - var ( - // Peer event channels. - peerCh = make(chan p2p.MeteredPeerEvent, eventBufferLimit) - // Subscribe to peer events. - subPeer = p2p.SubscribePeerEvent(peerCh) - ) - defer func() { - // Unsubscribe at the end. - subPeer.Unsubscribe() - }() + 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() db.peerLock.RLock() - //historyPeerLimiter := NewPeerLimiter(db.history.Network, false) - //historyFailedPeerLimiter := NewPeerLimiter(db.history.Network, true) - //db.peerLock.RUnlock() - //// Listen for events, and prepare the difference between two metering. - //diff := &NetworkMessage{ - // PeerBundles: make(map[string]*PeerBundle), - //} - //// Needed in order to keep the limit in the diff - //diffPeerLimiter := NewPeerLimiter(diff, false) - //diffFailedPeerLimiter := NewPeerLimiter(diff, true) + //historyMaintainer := NewPeerMaintainer(p2p.MeteredPeerLimit) + //historyHandshakeFailedMaintainer := NewPeerMaintainer(p2p.MeteredPeerLimit) + //diffMaintainer := NewPeerMaintainer(p2p.MeteredPeerLimit) + //diffHandshakeFailedMaintainer := NewPeerMaintainer(p2p.MeteredPeerLimit) for { select { case event := <-peerCh: fmt.Println(event) - //case event := <-connectCh: - // diffBundle := diff.getOrInitBundle(event.IP) - // diffBundle.Location = db.geodb.Location(event.IP) - // diffPeer := diffBundle.getOrInitPeer(event.ID) - // diffPeer.Connected = append(diffPeer.Connected, event.Connected) - // if first := len(diffPeer.Connected)-connectionLimit; first > 0 { - // diffPeer.Connected = diffPeer.Connected[first:] - // } - // diffPeer.ip = event.IP - // diffPeer.id = event.ID - // diffPeerLimiter.update(diffPeer) - //case event := <-disconnectCh: - // diffPeer := diff.getOrInitPeer(event.IP, event.ID) - // diffPeer.Disconnected = append(diffPeer.Disconnected, event.Time) - // if first := len(diffPeer.Connected)-connectionLimit; first > 0 { - // diffPeer.Connected = diffPeer.Connected[first:] - // } - // diffPeer.ip = event.IP - // diffPeer.id = event.ID - // diffPeerLimiter.update(diffPeer) - //case event := <-ingressCh: - // diffPeer := diff.getOrInitPeer(event.IP, event.ID) - // if len(diffPeer.Ingress) != 1 { - // diffPeer.Ingress = ChartEntries{&ChartEntry{Value: float64(event.Amount)}} - // } else { - // diffPeer.Ingress[0].Value = float64(event.Amount) - // } - // diffPeer.ip = event.IP - // diffPeer.id = event.ID - // diffPeerLimiter.update(diffPeer) - //case event := <-egressCh: - // diffPeer := diff.getOrInitPeer(event.IP, event.ID) - // if len(diffPeer.Egress) != 1 { - // diffPeer.Egress = ChartEntries{&ChartEntry{Value: float64(event.Amount)}} - // } else { - // diffPeer.Egress[0].Value = float64(event.Amount) - // } - // diffPeer.ip = event.IP - // diffPeer.id = event.ID - // diffPeerLimiter.update(diffPeer) - //case event := <-failedCh: - // diffBundle := diff.getOrInitBundle(event.IP) - // diffBundle.Location = db.geodb.Location(event.IP) - // id := fmt.Sprintf("peer_%d", atomic.AddInt64(&autoID, 1)) - // failedPeer := diffBundle.FailedPeers.getOrInit(id) - // failedPeer.Connected = []time.Time{event.Connected} - // failedPeer.Disconnected = []time.Time{event.Disconnected} - // failedPeer.ip = event.IP - // failedPeer.id = id - // diffFailedPeerLimiter.update(failedPeer) - //case <-ticker.C: - // now := time.Now() - // // Merge the diff with the history. - // db.peerLock.Lock() - // for ip, diffBundle := range diff.PeerBundles { - // historyBundle := db.history.Network.getOrInitBundle(ip) - // historyBundle.Location = diffBundle.Location - // for id, diffPeer := range diffBundle.Peers { - // historyPeer := historyBundle.getOrInitPeer(id) - // historyPeer.Connected = append(historyPeer.Connected, diffPeer.Connected...) - // if first := len(historyPeer.Connected)-connectionLimit; first > 0 { - // historyPeer.Connected = historyPeer.Connected[first:] - // } - // historyPeer.Disconnected = append(historyPeer.Disconnected, diffPeer.Disconnected...) - // if first := len(historyPeer.Disconnected)-connectionLimit; first > 0 { - // historyPeer.Disconnected = historyPeer.Disconnected[first:] - // } - // if len(diffPeer.Ingress) == 1 { - // diffPeer.Ingress[0].Time = now - // if historyPeer.Ingress == nil { - // historyPeer.Ingress = append(emptyChartEntries(now.Add(-db.config.Refresh), 3/*sampleLimit-1*/, db.config.Refresh), diffPeer.Ingress[0]) - // // The first message about a diffPeer should contain the whole list - // diffPeer.Ingress = historyPeer.Ingress - // } else { - // historyPeer.Ingress = append(historyPeer.Ingress, diffPeer.Ingress[0])[1:] - // } - // } - // if len(diffPeer.Egress) == 1 { - // diffPeer.Egress[0].Time = now - // if historyPeer.Egress == nil { - // historyPeer.Egress = append(emptyChartEntries(now.Add(-db.config.Refresh), 3/*sampleLimit-1*/, db.config.Refresh), diffPeer.Egress[0]) - // // The first message about a diffPeer should contain the whole list - // diffPeer.Egress = historyPeer.Egress - // } else { - // historyPeer.Egress = append(historyPeer.Egress, diffPeer.Egress[0])[1:] - // } - // } - // historyPeer.ip = diffPeer.ip - // historyPeer.id = diffPeer.id - // historyPeerLimiter.update(historyPeer) - // } - // for id, diffFailedPeer := range diffBundle.FailedPeers { - // historyFailedPeer := historyBundle.getOrInitFailedPeer(id) - // historyFailedPeer.Connected = diffFailedPeer.Connected - // historyFailedPeer.Disconnected = diffFailedPeer.Disconnected - // historyFailedPeer.ip = diffFailedPeer.ip - // historyFailedPeer.id = diffFailedPeer.id - // historyFailedPeerLimiter.update(historyFailedPeer) - // } - // } - // //for elem := historyPeerLimiter.l.Front(); elem != nil; elem = elem.Next() { - // // s, _ := json.MarshalIndent(elem.Value, "", " ") - // // fmt.Println(string(s)) - // //} - // //fmt.Println() - // db.peerLock.Unlock() - // - // //s, _ := json.MarshalIndent(deepcopy.Copy(diff), "", " ") - // //fmt.Println(string(s)) - // //fmt.Println() - // // Send the diff to the clients. - // db.sendToAll(&Message{Network: deepcopy.Copy(diff).(*NetworkMessage)}) - // - // // Prepare for the next metering, clear the diff variable. - // diffPeerLimiter.clear() - // diffFailedPeerLimiter.clear() - // diff = &NetworkMessage{ - // PeerBundles: make(map[string]*PeerBundle), - // } + //diffMaintainer.Update(event.IP.String(), event.ID) case err := <-subPeer.Err(): log.Warn("Peer subscription error", "err", err) return @@ -250,4 +229,3 @@ func (db *Dashboard) collectPeerData() { } } } - diff --git a/p2p/metrics.go b/p2p/metrics.go index 1af13df5a9..86401379df 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -37,10 +37,7 @@ const ( MetricsOutboundConnects = "p2p/OutboundConnects" // Name for the registered outbound connects meter MetricsOutboundTraffic = "p2p/OutboundTraffic" // Name for the registered outbound traffic meter - MetricsRegistryIngressPrefix = MetricsInboundTraffic + "/" - MetricsRegistryEgressPrefix = MetricsOutboundTraffic + "/" - - MeteredPeerLimit = 1024 + MeteredPeerLimit = 1024 // This amount of peers are individually metered ) var ( @@ -49,11 +46,11 @@ var ( egressConnectMeter = metrics.NewRegisteredMeter(MetricsOutboundConnects, nil) // Meter counting the egress connections egressTrafficMeter = metrics.NewRegisteredMeter(MetricsOutboundTraffic, nil) // Meter metering the cumulative egress traffic - PeerIngressRegistry = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, MetricsRegistryIngressPrefix) // Registry containing the peer ingress - PeerEgressRegistry = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, MetricsRegistryEgressPrefix) // Registry containing the peer egress + PeerIngressRegistry = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, MetricsInboundTraffic+"/") // Registry containing the peer ingress + PeerEgressRegistry = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, MetricsOutboundTraffic+"/") // Registry containing the peer egress - metricsFeed event.Feed // Event feed for peer metrics - meteredPeerCount uint64 // Actually stored peer connection count + meteredPeerFeed event.Feed // Event feed for peer metrics + meteredPeerCount int32 // Actually stored peer connection count ) // MeteredPeerEventType is the type of peer events emitted by a metered connection. @@ -72,19 +69,20 @@ const ( PeerHandshakeFailed ) -// MeteredPeerEvent is an event emitted when peers connect or disconnect +// MeteredPeerEvent is an event emitted when peers connect or disconnect. type MeteredPeerEvent struct { Type MeteredPeerEventType // Type of peer event IP net.IP // IP address of the peer ID string // NodeID of the peer Elapsed time.Duration // Time elapsed between the connection and the handshake/disconnection - Ingress uint64 // Ingress count in the moment of disconnection - Egress uint64 // Egress count in the moment of disconnection + Ingress uint64 // Ingress count at the moment of the event + Egress uint64 // Egress count at the moment of the event } -// SubscribePeerEvent registers a subscription of PeerEvent -func SubscribePeerEvent(ch chan<- MeteredPeerEvent) event.Subscription { - return metricsFeed.Subscribe(ch) +// SubscribeMeteredPeerEvent registers a subscription for peer life-cycle events +// if metrics collection is enabled. +func SubscribeMeteredPeerEvent(ch chan<- MeteredPeerEvent) event.Subscription { + return meteredPeerFeed.Subscribe(ch) } // meteredConn is a wrapper around a net.Conn that meters both the @@ -95,6 +93,7 @@ type meteredConn struct { connected time.Time // Connection time of the peer ip net.IP // IP address of the peer id string // NodeID of the peer + metered bool // Checks if the peer is metered ingressMeter metrics.Meter // Meter for the read bytes of the peer egressMeter metrics.Meter // Meter for the written bytes of the peer @@ -103,8 +102,8 @@ type meteredConn struct { // newMeteredConn creates a new metered connection, bumps the ingress or egress // connection meter and also increases the metered peer count. If the metrics -// system is disabled, the IP address is unspecified or the metered peer count -// reached the limit, this function returns the original object. +// system is disabled or the IP address is unspecified, this function returns +// the original object. func newMeteredConn(conn net.Conn, ingress bool, ip net.IP) net.Conn { // Short circuit if metrics are disabled if !metrics.Enabled { @@ -114,12 +113,6 @@ func newMeteredConn(conn net.Conn, ingress bool, ip net.IP) net.Conn { log.Warn("Peer IP is unspecified") return conn } - if atomic.LoadUint64(&meteredPeerCount) >= MeteredPeerLimit { - log.Warn("Metered peer count reached the limit") - return conn - } - // Increment the metered peer count - atomic.AddUint64(&meteredPeerCount, 1) // Bump the connection counters and wrap the connection if ingress { ingressConnectMeter.Mark(1) @@ -139,7 +132,7 @@ func (c *meteredConn) Read(b []byte) (n int, err error) { n, err = c.Conn.Read(b) ingressTrafficMeter.Mark(int64(n)) c.lock.RLock() - if c.ingressMeter != nil { + if c.metered { c.ingressMeter.Mark(int64(n)) } c.lock.RUnlock() @@ -152,7 +145,7 @@ func (c *meteredConn) Write(b []byte) (n int, err error) { n, err = c.Conn.Write(b) egressTrafficMeter.Mark(int64(n)) c.lock.RLock() - if c.egressMeter != nil { + if c.metered { c.egressMeter.Mark(int64(n)) } c.lock.RUnlock() @@ -160,40 +153,51 @@ func (c *meteredConn) Write(b []byte) (n int, err error) { } // handshakeDone is called when a peer handshake is done. Registers the peer to -// the ingress and the egress traffic registries using the peer's IP and NodeID, -// also emits connect event. -func (c *meteredConn) handshakeDone(id discover.NodeID) { +// the ingress and the egress traffic registries using the peer's IP and NodeID +// if the metered peer count didn't reach the limit, also emits connect event. +func (c *meteredConn) handshakeDone(nodeID discover.NodeID) { + if atomic.LoadInt32(&meteredPeerCount) >= MeteredPeerLimit { + log.Warn("Metered peer count reached the limit") + return + } + // Increment the metered peer count + atomic.AddInt32(&meteredPeerCount, 1) + id := nodeID.String() + key := fmt.Sprintf("%s/%s", c.ip, id) c.lock.Lock() - c.id = id.String() - key := fmt.Sprintf("%s/%s", c.ip, c.id) + c.id, c.metered = id, true c.ingressMeter = metrics.NewRegisteredMeter(key, PeerIngressRegistry) c.egressMeter = metrics.NewRegisteredMeter(key, PeerEgressRegistry) c.lock.Unlock() - metricsFeed.Send(MeteredPeerEvent{ + meteredPeerFeed.Send(MeteredPeerEvent{ Type: PeerConnected, IP: c.ip, - ID: id.String(), - Elapsed: time.Now().Sub(c.connected), + ID: id, + Elapsed: time.Since(c.connected), }) } // Close delegates a close operation to the underlying connection, unregisters // the peer from the traffic registries and emits close event. func (c *meteredConn) Close() error { - // Decrement the metered peer count - atomic.AddUint64(&meteredPeerCount, ^uint64(0)) - + err := c.Conn.Close() c.lock.RLock() - // If the peer disconnects before the handshake - if c.id == "" { + if !c.metered { c.lock.RUnlock() - metricsFeed.Send(MeteredPeerEvent{ + return err + } + // Decrement the metered peer count + atomic.AddInt32(&meteredPeerCount, -1) + if c.id == "" { + // If the peer disconnects before the handshake + c.lock.RUnlock() + meteredPeerFeed.Send(MeteredPeerEvent{ Type: PeerHandshakeFailed, IP: c.ip, - Elapsed: time.Now().Sub(c.connected), + Elapsed: time.Since(c.connected), }) - return c.Conn.Close() + return err } id, ingress, egress := c.id, uint64(c.ingressMeter.Count()), uint64(c.egressMeter.Count()) c.lock.RUnlock() @@ -203,12 +207,12 @@ func (c *meteredConn) Close() error { PeerIngressRegistry.Unregister(key) PeerEgressRegistry.Unregister(key) - metricsFeed.Send(MeteredPeerEvent{ + meteredPeerFeed.Send(MeteredPeerEvent{ Type: PeerDisconnected, IP: c.ip, ID: id, Ingress: ingress, Egress: egress, }) - return c.Conn.Close() + return err }