chore: refactor connection manager and registrar

Author: vasco-santos

doc/API.md

@@ -11,12 +11,12 @@
   * [`handle`](#handle)
   * [`unhandle`](#unhandle)
   * [`ping`](#ping)
-  * [`peerRouting.findPeer`](#peerroutingfindpeer)
   * [`contentRouting.findProviders`](#contentroutingfindproviders)
   * [`contentRouting.provide`](#contentroutingprovide)
   * [`contentRouting.put`](#contentroutingput)
   * [`contentRouting.get`](#contentroutingget)
   * [`contentRouting.getMany`](#contentroutinggetmany)
+  * [`peerRouting.findPeer`](#peerroutingfindpeer)
   * [`peerStore.addressBook.add`](#peerstoreaddressbookadd)
   * [`peerStore.addressBook.delete`](#peerstoreaddressbookdelete)
   * [`peerStore.addressBook.get`](#peerstoreaddressbookget)
@@ -34,14 +34,17 @@
   * [`pubsub.publish`](#pubsubpublish)
   * [`pubsub.subscribe`](#pubsubsubscribe)
   * [`pubsub.unsubscribe`](#pubsubunsubscribe)
+  * [`connectionManager.get`](#connectionmanagerget)
   * [`connectionManager.setPeerValue`](#connectionmanagersetpeervalue)
+  * [`connectionManager.size`](#connectionmanagersize)
   * [`metrics.global`](#metricsglobal)
   * [`metrics.peers`](#metricspeers)
   * [`metrics.protocols`](#metricsprotocols)
   * [`metrics.forPeer`](#metricsforpeer)
   * [`metrics.forProtocol`](#metricsforprotocol)
 * [Events](#events)
   * [`libp2p`](#libp2p)
+  * [`libp2p.connectionManager`](#libp2pconnectionmanager)
   * [`libp2p.peerStore`](#libp2ppeerStore)
 * [Types](#types)
   * [`Stats`](#stats)
@@ -999,6 +1002,28 @@ const handler = (msg) => {
 libp2p.pubsub.unsubscribe(topic, handler)
 ```
 
+### connectionManager.get
+
+Get a connection with a given peer, if it exists.
+
+#### Parameters
+
+| Name | Type | Description |
+|------|------|-------------|
+| peerId | [`PeerId`][peer-id] | The peer to find |
+
+#### Returns
+
+| Type | Description |
+|------|-------------|
+| [`Connection`][connection] | Connection with the given peer |
+
+#### Example
+
+```js
+libp2p.connectionManager.get(peerId)
+```
+
 ### connectionManager.setPeerValue
 
 Enables users to change the value of certain peers in a range of 0 to 1. Peers with the lowest values will have their Connections pruned first, if any Connection Manager limits are exceeded. See [./CONFIGURATION.md#configuring-connection-manager](./CONFIGURATION.md#configuring-connection-manager) for details on how to configure these limits.
@@ -1025,6 +1050,17 @@ libp2p.connectionManager.setPeerValue(highPriorityPeerId, 1)
 libp2p.connectionManager.setPeerValue(lowPriorityPeerId, 0)
 ```
 
+### connectionManager.size
+
+Getter for obtaining the current number of open connections.
+
+#### Example
+
+```js
+libp2p.connectionManager.size
+// 10
+```
+
 ### metrics.global
 
 A [`Stats`](#stats) object of tracking the global bandwidth of the libp2p node.
@@ -1126,6 +1162,8 @@ unless they are performing a specific action. See [peer discovery and auto dial]
 
 - `peer`: instance of [`PeerId`][peer-id]
 
+### libp2p.connectionManager
+
 #### A new connection to a peer has been opened
 
 This event will be triggered anytime a new Connection is established to another peer.

src/circuit/circuit/hop.js

@@ -41,7 +41,7 @@ module.exports.handleHop = async function handleHop ({
   // Get the connection to the destination (stop) peer
   const destinationPeer = new PeerId(request.dstPeer.id)
 
-  const destinationConnection = circuit._registrar.getConnection(destinationPeer)
+  const destinationConnection = circuit._connectionManager.get(destinationPeer)
   if (!destinationConnection && !circuit._options.hop.active) {
     log('HOP request received but we are not connected to the destination peer')
     return streamHandler.end({

src/circuit/index.js

@@ -29,6 +29,7 @@ class Circuit {
   constructor ({ libp2p, upgrader }) {
     this._dialer = libp2p.dialer
     this._registrar = libp2p.registrar
+    this._connectionManager = libp2p.connectionManager
     this._upgrader = upgrader
     this._options = libp2p._config.relay
     this.addresses = libp2p.addresses
@@ -107,7 +108,7 @@ class Circuit {
     const destinationPeer = PeerId.createFromCID(destinationAddr.getPeerId())
 
     let disconnectOnFailure = false
-    let relayConnection = this._registrar.getConnection(relayPeer)
+    let relayConnection = this._connectionManager.get(relayPeer)
     if (!relayConnection) {
       relayConnection = await this._dialer.connectToPeer(relayAddr, options)
       disconnectOnFailure = true

src/connection-manager/index.js

@@ -6,6 +6,11 @@ const LatencyMonitor = require('latency-monitor').default
 const debug = require('debug')('libp2p:connection-manager')
 const retimer = require('retimer')
 
+const { EventEmitter } = require('events')
+
+const PeerId = require('peer-id')
+const { Connection } = require('libp2p-interfaces/src/connection')
+
 const {
   ERR_INVALID_PARAMETERS
 } = require('../errors')
@@ -22,7 +27,12 @@ const defaultOptions = {
   defaultPeerValue: 1
 }
 
-class ConnectionManager {
+/**
+ * Responsible for managing known connections.
+ * @fires ConnectionManager#peer:connect Emitted when a new peer is connected.
+ * @fires ConnectionManager#peer:disconnect Emitted when a known peer supports a different set of protocols.
+ */
+class ConnectionManager extends EventEmitter {
   /**
    * @constructor
    * @param {Libp2p} libp2p
@@ -38,9 +48,11 @@ class ConnectionManager {
    * @param {Number} options.defaultPeerValue The value of the peer. Default=1
    */
   constructor (libp2p, options) {
+    super()
+
     this._libp2p = libp2p
-    this._registrar = libp2p.registrar
     this._peerId = libp2p.peerId.toB58String()
+
     this._options = mergeOptions.call({ ignoreUndefined: true }, defaultOptions, options)
     if (this._options.maxConnections < this._options.minConnections) {
       throw errcode(new Error('Connection Manager maxConnections must be greater than minConnections'), ERR_INVALID_PARAMETERS)
@@ -50,12 +62,30 @@ class ConnectionManager {
 
     this._metrics = libp2p.metrics
 
+    /**
+     * Map of peer identifiers to their peer value for pruning connections.
+     * @type {Map<string, number>}
+     */
     this._peerValues = new Map()
-    this._connections = new Map()
+
+    /**
+     * Map of connections per peer
+     * @type {Map<string, Array<conn>>}
+     */
+    this.connections = new Map()
+
     this._timer = null
     this._checkMetrics = this._checkMetrics.bind(this)
   }
 
+  /**
+   * Get current number of open connections.
+   */
+  get size () {
+    return Array.from(this.connections.values())
+      .reduce((accumulator, value) => accumulator + value.length, 0)
+  }
+
   /**
    * Starts the Connection Manager. If Metrics are not enabled on libp2p
    * only event loop and connection limits will be monitored.
@@ -77,13 +107,33 @@ class ConnectionManager {
 
   /**
    * Stops the Connection Manager
+   * @async
    */
-  stop () {
+  async stop () {
     this._timer && this._timer.clear()
     this._latencyMonitor && this._latencyMonitor.removeListener('data', this._onLatencyMeasure)
+
+    await this._close()
     debug('stopped')
   }
 
+  /**
+   * Cleans up the connections
+   * @async
+   */
+  async _close () {
+    // Close all connections we're tracking
+    const tasks = []
+    for (const connectionList of this.connections.values()) {
+      for (const connection of connectionList) {
+        tasks.push(connection.close())
+      }
+    }
+
+    await tasks
+    this.connections.clear()
+  }
+
   /**
    * Sets the value of the given peer. Peers with lower values
    * will be disconnected first.
@@ -122,21 +172,65 @@ class ConnectionManager {
    * @param {Connection} connection
    */
   onConnect (connection) {
-    const peerId = connection.remotePeer.toB58String()
-    this._connections.set(connection.id, connection)
-    if (!this._peerValues.has(peerId)) {
-      this._peerValues.set(peerId, this._options.defaultPeerValue)
+    if (!Connection.isConnection(connection)) {
+      throw errcode(new Error('conn must be an instance of interface-connection'), ERR_INVALID_PARAMETERS)
     }
-    this._checkLimit('maxConnections', this._connections.size)
+
+    const peerId = connection.remotePeer
+    const peerIdStr = peerId.toB58String()
+    const storedConn = this.connections.get(peerIdStr)
+
+    if (storedConn) {
+      storedConn.push(connection)
+    } else {
+      this.connections.set(peerIdStr, [connection])
+      this.emit('peer:connect', peerId)
+    }
+
+    if (!this._peerValues.has(peerIdStr)) {
+      this._peerValues.set(peerIdStr, this._options.defaultPeerValue)
+    }
+
+    this._checkLimit('maxConnections', this.size)
   }
 
   /**
    * Removes the connection from tracking
    * @param {Connection} connection
    */
   onDisconnect (connection) {
-    this._connections.delete(connection.id)
-    this._peerValues.delete(connection.remotePeer.toB58String())
+    const peerId = connection.remotePeer
+    const peerIdStr = peerId.toB58String()
+    let storedConn = this.connections.get(peerIdStr)
+
+    if (storedConn && storedConn.length > 1) {
+      storedConn = storedConn.filter((conn) => conn.id !== connection.id)
+      this.connections.set(peerIdStr, storedConn)
+    } else if (storedConn) {
+      this.connections.delete(peerIdStr)
+      this._peerValues.delete(connection.remotePeer.toB58String())
+      this.emit('peer:disconnect', peerId)
+    }
+  }
+
+  /**
+   * Get a connection with a peer.
+   * @param {PeerId} peerId
+   * @returns {Connection}
+   */
+  get (peerId) {
+    if (!PeerId.isPeerId(peerId)) {
+      throw errcode(new Error('peerId must be an instance of peer-id'), ERR_INVALID_PARAMETERS)
+    }
+
+    const id = peerId.toB58String()
+    const connections = this.connections.get(id)
+
+    // Return the first, open connection
+    if (connections) {
+      return connections.find(connection => connection.stat.status === 'open')
+    }
+    return null
   }
 
   /**
@@ -169,17 +263,17 @@ class ConnectionManager {
    * @private
    */
   _maybeDisconnectOne () {
-    if (this._options.minConnections < this._connections.size) {
+    if (this._options.minConnections < this.connections.size) {
       const peerValues = Array.from(this._peerValues).sort(byPeerValue)
       debug('%s: sorted peer values: %j', this._peerId, peerValues)
       const disconnectPeer = peerValues[0]
       if (disconnectPeer) {
         const peerId = disconnectPeer[0]
         debug('%s: lowest value peer is %s', this._peerId, peerId)
         debug('%s: closing a connection to %j', this._peerId, peerId)
-        for (const connection of this._connections.values()) {
-          if (connection.remotePeer.toB58String() === peerId) {
-            connection.close()
+        for (const connections of this.connections.values()) {
+          if (connections[0].remotePeer.toB58String() === peerId) {
+            connections[0].close()
             break
           }
         }

src/identify/index.js

@@ -45,16 +45,23 @@ class IdentifyService {
   /**
    * @constructor
    * @param {object} options
-   * @param {Registrar} options.registrar
+   * @param {PeerStore} options.peerStore
+   * @param {ConnectionManager} options.connectionManager
    * @param {Map<string, handler>} options.protocols A reference to the protocols we support
    * @param {PeerId} options.peerId The peer running the identify service
    * @param {{ listen: Array<Multiaddr>}} options.addresses The peer aaddresses
    */
   constructor (options) {
     /**
-     * @property {Registrar}
+     * @property {PeerStore}
      */
-    this.registrar = options.registrar
+    this.peerStore = options.peerStore
+
+    /**
+     * @property {ConnectionManager}
+     */
+    this.connectionManager = options.connectionManager
+
     /**
      * @property {PeerId}
      */
@@ -103,7 +110,7 @@ class IdentifyService {
     const connections = []
     let connection
     for (const peer of peerStore.peers.values()) {
-      if (peer.protocols.includes(MULTICODEC_IDENTIFY_PUSH) && (connection = this.registrar.getConnection(peer.id))) {
+      if (peer.protocols.includes(MULTICODEC_IDENTIFY_PUSH) && (connection = this.connectionManager.get(peer.id))) {
         connections.push(connection)
       }
     }
@@ -159,8 +166,8 @@ class IdentifyService {
     observedAddr = IdentifyService.getCleanMultiaddr(observedAddr)
 
     // Update peers data in PeerStore
-    this.registrar.peerStore.addressBook.set(id, listenAddrs.map((addr) => multiaddr(addr)))
-    this.registrar.peerStore.protoBook.set(id, protocols)
+    this.peerStore.addressBook.set(id, listenAddrs.map((addr) => multiaddr(addr)))
+    this.peerStore.protoBook.set(id, protocols)
 
     // TODO: Track our observed address so that we can score it
     log('received observed address of %s', observedAddr)
@@ -244,13 +251,13 @@ class IdentifyService {
     // Update peers data in PeerStore
     const id = connection.remotePeer
     try {
-      this.registrar.peerStore.addressBook.set(id, message.listenAddrs.map((addr) => multiaddr(addr)))
+      this.peerStore.addressBook.set(id, message.listenAddrs.map((addr) => multiaddr(addr)))
     } catch (err) {
       return log.error('received invalid listen addrs', err)
     }
 
     // Update the protocols
-    this.registrar.peerStore.protoBook.set(id, message.protocols)
+    this.peerStore.protoBook.set(id, message.protocols)
   }
 }