dashboard, p2p: experiment 3

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 (
-const trafficEventBufferLimit = p2p.MeteredPeerLimit
+	eventBufferLimit = 128 // Maximum number of buffered peer events
-const connectionLimit = 100
+	connectionLimit  = 16
+)
 
 var autoID int64
 
-type peerLimiter struct {
+// maintainedPeer is the element of the peer maintainer's linked list.
-	underlying *NetworkMessage
+// Similarly to an ordinary linked list element it knows the previous
-	failed bool
+// and the next elements, and also has a pointer to its parent map in
-	l *list.List
+// 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 {
+// idContainer contains the NodeIDs belonging to an IP address.
-	return &peerLimiter{l: list.New(), underlying: underlying, failed: failed}
+// 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) {
+// update moves the list element belonging to the given ID to the end,
-	return
+// or inserts a new element to the end of the list if the given ID didn't
-	if peer.element == nil {
+// appear yet in the container. Returns the IP and the ID of the removed
-		peer.element = pl.l.PushBack(peer)
+// 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 {
+	if maintainer.len > maintainer.limit {
-		pl.remove(pl.l.Front())
+		first := maintainer.remove(maintainer.root.next)
+		return &removedPeer{
+			ip: first.parent.ip,
+			id: first.id,
+		}
+	}
+	return nil
+}
+
+// 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
 	}
 }
 
-func (pl *peerLimiter) remove(e *list.Element) {
+// PeerMaintainer is an abstract layer above the metered peer container,
-	return
+// maintaining the peers. i.e. sorting them based on their activity and
-	elem := pl.l.Remove(e)
+// removing the oldest inactive ones when their count reaches the limit.
-	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)
-		}
-	}
-}
-
-func (pl *peerLimiter) clear() {
-	for pl.l.Front() != nil {
-		pl.remove(pl.l.Front())
-	}
-}
 //
-//func tail(arr []time.Time) []time.Time {
+// Consists of a map of maps which represent the peers grouped by the IP
-//	if first := len(arr)-connectionLimit; first > 0 {
+// address then by the NodeID. The elements on the bottom of the tree are
-//		return arr[first:]
+// 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
-//	return arr
+// 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 (
+	peerCh := make(chan p2p.MeteredPeerEvent, eventBufferLimit) // Peer event channel.
-		// Peer event channels.
+	subPeer := p2p.SubscribeMeteredPeerEvent(peerCh)            // Subscribe to peer events.
-		peerCh    = make(chan p2p.MeteredPeerEvent, eventBufferLimit)
+	defer subPeer.Unsubscribe()                                 // Unsubscribe at the end.
-		// Subscribe to peer events.
-		subPeer    = p2p.SubscribePeerEvent(peerCh)
-	)
-	defer func() {
-		// Unsubscribe at the end.
-		subPeer.Unsubscribe()
-	}()
 
 	ticker := time.NewTicker(db.config.Refresh)
 	defer ticker.Stop()
 
 	db.peerLock.RLock()
-	//historyPeerLimiter := NewPeerLimiter(db.history.Network, false)
+	//historyMaintainer := NewPeerMaintainer(p2p.MeteredPeerLimit)
-	//historyFailedPeerLimiter := NewPeerLimiter(db.history.Network, true)
+	//historyHandshakeFailedMaintainer := NewPeerMaintainer(p2p.MeteredPeerLimit)
-	//db.peerLock.RUnlock()
+	//diffMaintainer := NewPeerMaintainer(p2p.MeteredPeerLimit)
-	//// Listen for events, and prepare the difference between two metering.
+	//diffHandshakeFailedMaintainer := NewPeerMaintainer(p2p.MeteredPeerLimit)
-	//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)
 	for {
 		select {
 		case event := <-peerCh:
 			fmt.Println(event)
-		//case event := <-connectCh:
+			//diffMaintainer.Update(event.IP.String(), event.ID)
-		//	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),
-		//	}
 		case err := <-subPeer.Err():
 			log.Warn("Peer subscription error", "err", err)
 			return
@@ -250,4 +229,3 @@ func (db *Dashboard) collectPeerData() {
 		}
 	}
 }
-

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 + "/"
+	MeteredPeerLimit = 1024 // This amount of peers are individually metered
-	MetricsRegistryEgressPrefix  = MetricsOutboundTraffic + "/"
-
-	MeteredPeerLimit = 1024
 )
 
 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
+	PeerIngressRegistry = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, MetricsInboundTraffic+"/")  // Registry containing the peer ingress
-	PeerEgressRegistry  = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, MetricsRegistryEgressPrefix)  // Registry containing the peer egress
+	PeerEgressRegistry  = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, MetricsOutboundTraffic+"/") // Registry containing the peer egress
 
-	metricsFeed      event.Feed // Event feed for peer metrics
+	meteredPeerFeed  event.Feed // Event feed for peer metrics
-	meteredPeerCount uint64     // Actually stored peer connection count
+	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
+	Ingress uint64               // Ingress count at the moment of the event
-	Egress  uint64               // Egress count in the moment of disconnection
+	Egress  uint64               // Egress count at the moment of the event
 }
 
-// SubscribePeerEvent registers a subscription of PeerEvent
+// SubscribeMeteredPeerEvent registers a subscription for peer life-cycle events
-func SubscribePeerEvent(ch chan<- MeteredPeerEvent) event.Subscription {
+// if metrics collection is enabled.
-	return metricsFeed.Subscribe(ch)
+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
+// system is disabled or the IP address is unspecified, this function returns
-// reached the limit, this function returns the original object.
+// 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,
+// the ingress and the egress traffic registries using the peer's IP and NodeID
-// also emits connect event.
+// if the metered peer count didn't reach the limit, also emits connect event.
-func (c *meteredConn) handshakeDone(id discover.NodeID) {
+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()
+	c.id, c.metered = id, true
-	key := fmt.Sprintf("%s/%s", c.ip, c.id)
 	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(),
+		ID:      id,
-		Elapsed: time.Now().Sub(c.connected),
+		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
+	err := c.Conn.Close()
-	atomic.AddUint64(&meteredPeerCount, ^uint64(0))
-
 	c.lock.RLock()
-	// If the peer disconnects before the handshake
+	if !c.metered {
-	if c.id == "" {
 		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
 }