feat: support a priority queue for dials (libp2p#325)

Author: jacobheun

README.md

@@ -153,6 +153,16 @@ works like dial, but calls back with a [Connection State Machine](#connection-st
 Connection state machines emit a number of events that can be used to determine the current state of the connection
 and to received the underlying connection that can be used to transfer data.
 
+### `switch.dialer.connect(peer, options, callback)`
+
+a low priority dial to the provided peer. Calls to `dial` and `dialFSM` will take priority. This should be used when an application only wishes to establish connections to new peers, such as during peer discovery when there is a low peer count. Currently, anything greater than the HIGH_PRIORITY (10) will be placed into the cold call queue, and anything less than or equal to the HIGH_PRIORITY will be added to the normal queue.
+
+- `peer`: can be an instance of [PeerInfo][], [PeerId][] or [multiaddr][]
+- `options`: Optional
+- `options.priority`: Number of the priority of the dial, defaults to 20.
+- `options.useFSM`: Boolean of whether or not to callback with a [Connection State Machine](#connection-state-machine)
+- `callback`: Function with signature `function (err, connFSM) {}` where `connFSM` is a [Connection State Machine](#connection-state-machine)
+
 ##### Events
 - `error`:  emitted whenever a fatal error occurs with the connection; the error will be emitted.
 - `error:upgrade_failed`: emitted whenever the connection fails to upgrade with a muxer, this is not fatal.
@@ -187,7 +197,17 @@ Emitted when the switch encounters an error.
 
 ### `switch.on('peer-mux-closed', (peer) => {})`
 
-- `peer`: is instance of [PeerInfo][] that has info of the peer we have just closed a muxed connection.
+- `peer`: is instance of [PeerInfo][] that has info of the peer we have just closed a muxed connection with.
+
+### `switch.on('connection:start', (peer) => {})`
+This will be triggered anytime a new connection is created.
+
+- `peer`: is instance of [PeerInfo][] that has info of the peer we have just started a connection with.
+
+### `switch.on('connection:end', (peer) => {})`
+This will be triggered anytime an existing connection, regardless of state, is removed from the switch's internal connection tracking.
+
+- `peer`: is instance of [PeerInfo][] that has info of the peer we have just closed a connection with.
 
 ### `switch.on('start', () => {})`
 

src/connection/base.js

@@ -80,6 +80,7 @@ class BaseConnection extends EventEmitter {
    * @returns {void}
    */
   _onDisconnected () {
+    this.switch.connection.remove(this)
     this.log('disconnected from %s', this.theirB58Id)
     this.emit('close')
     this.removeAllListeners()

src/connection/index.js

@@ -269,8 +269,6 @@ class ConnectionFSM extends BaseConnection {
   _onDisconnecting () {
     this.log('disconnecting from %s', this.theirB58Id, Boolean(this.muxer))
 
-    this.switch.connection.remove(this)
-
     delete this.switch.conns[this.theirB58Id]
 
     let tasks = []

src/connection/manager.js

@@ -33,7 +33,12 @@ class ConnectionManager {
     // Only add it if it's not there
     if (!this.get(connection)) {
       this.connections[connection.theirB58Id].push(connection)
-      this.switch.emit('peer-mux-established', connection.theirPeerInfo)
+      this.switch.emit('connection:start', connection.theirPeerInfo)
+      if (connection.getState() === 'MUXED') {
+        this.switch.emit('peer-mux-established', connection.theirPeerInfo)
+      } else {
+        connection.once('muxed', () => this.switch.emit('peer-mux-established', connection.theirPeerInfo))
+      }
     }
   }
 
@@ -81,8 +86,10 @@ class ConnectionManager {
   remove (connection) {
     // No record of the peer, disconnect it
     if (!this.connections[connection.theirB58Id]) {
-      connection.theirPeerInfo.disconnect()
-      this.switch.emit('peer-mux-closed', connection.theirPeerInfo)
+      if (connection.theirPeerInfo) {
+        connection.theirPeerInfo.disconnect()
+        this.switch.emit('peer-mux-closed', connection.theirPeerInfo)
+      }
       return
     }
 
@@ -99,6 +106,9 @@ class ConnectionManager {
       connection.theirPeerInfo.disconnect()
       this.switch.emit('peer-mux-closed', connection.theirPeerInfo)
     }
+
+    // A tracked connection was closed, let the world know
+    this.switch.emit('connection:end', connection.theirPeerInfo)
   }
 
   /**

src/constants.js

@@ -6,5 +6,7 @@ module.exports = {
   DIAL_TIMEOUT: 30e3, // How long in ms a dial attempt is allowed to take
   MAX_COLD_CALLS: 50, // How many dials w/o protocols that can be queued
   MAX_PARALLEL_DIALS: 100, // Maximum allowed concurrent dials
-  QUARTER_HOUR: 15 * 60e3
+  QUARTER_HOUR: 15 * 60e3,
+  PRIORITY_HIGH: 10,
+  PRIORITY_LOW: 20
 }

src/dialer/index.js

@@ -6,7 +6,9 @@ const {
   BLACK_LIST_ATTEMPTS,
   BLACK_LIST_TTL,
   MAX_COLD_CALLS,
-  MAX_PARALLEL_DIALS
+  MAX_PARALLEL_DIALS,
+  PRIORITY_HIGH,
+  PRIORITY_LOW
 } = require('../constants')
 
 module.exports = function (_switch) {
@@ -19,7 +21,7 @@ module.exports = function (_switch) {
    * @param {DialRequest} dialRequest
    * @returns {void}
    */
-  function _dial ({ peerInfo, protocol, useFSM, callback }) {
+  function _dial ({ peerInfo, protocol, options, callback }) {
     if (typeof protocol === 'function') {
       callback = protocol
       protocol = null
@@ -32,7 +34,7 @@ module.exports = function (_switch) {
     }
 
     // Add it to the queue, it will automatically get executed
-    dialQueueManager.add({ peerInfo, protocol, useFSM, callback })
+    dialQueueManager.add({ peerInfo, protocol, options, callback })
   }
 
   /**
@@ -64,14 +66,33 @@ module.exports = function (_switch) {
     dialQueueManager.clearBlacklist(peerInfo)
   }
 
+  /**
+   * Attempts to establish a connection to the given `peerInfo` at
+   * a lower priority than a standard dial.
+   * @param {PeerInfo} peerInfo
+   * @param {object} options
+   * @param {boolean} options.useFSM Whether or not to return a `ConnectionFSM`. Defaults to false.
+   * @param {number} options.priority Lowest priority goes first. Defaults to 20.
+   * @param {function(Error, Connection)} callback
+   */
+  function connect (peerInfo, options, callback) {
+    if (typeof options === 'function') {
+      callback = options
+      options = null
+    }
+    options = { useFSM: false, priority: PRIORITY_LOW, ...options }
+    _dial({ peerInfo, protocol: null, options, callback })
+  }
+
   /**
    * Adds the dial request to the queue for the given `peerInfo`
+   * The request will be added with a high priority (10).
    * @param {PeerInfo} peerInfo
    * @param {string} protocol
    * @param {function(Error, Connection)} callback
    */
   function dial (peerInfo, protocol, callback) {
-    _dial({ peerInfo, protocol, useFSM: false, callback })
+    _dial({ peerInfo, protocol, options: { useFSM: false, priority: PRIORITY_HIGH }, callback })
   }
 
   /**
@@ -82,10 +103,11 @@ module.exports = function (_switch) {
    * @param {function(Error, ConnectionFSM)} callback
    */
   function dialFSM (peerInfo, protocol, callback) {
-    _dial({ peerInfo, protocol, useFSM: true, callback })
+    _dial({ peerInfo, protocol, options: { useFSM: true, priority: PRIORITY_HIGH }, callback })
   }
 
   return {
+    connect,
     dial,
     dialFSM,
     clearBlacklist,

src/dialer/queue.js

@@ -13,7 +13,9 @@ log.error = debug('libp2p:switch:dial:error')
  * @typedef {Object} DialRequest
  * @property {PeerInfo} peerInfo - The peer to dial to
  * @property {string} [protocol] - The protocol to create a stream for
- * @property {boolean} useFSM - If `callback` should return a ConnectionFSM
+ * @property {object} options
+ * @property {boolean} options.useFSM - If `callback` should return a ConnectionFSM
+ * @property {number} options.priority - The priority of the dial
  * @property {function(Error, Connection|ConnectionFSM)} callback
  */
 

src/dialer/queueManager.js

@@ -5,7 +5,7 @@ const Queue = require('./queue')
 const { DIAL_ABORTED } = require('../errors')
 const nextTick = require('async/nextTick')
 const retimer = require('retimer')
-const { QUARTER_HOUR } = require('../constants')
+const { QUARTER_HOUR, PRIORITY_HIGH } = require('../constants')
 const debug = require('debug')
 const log = debug('libp2p:switch:dial:manager')
 const noop = () => {}
@@ -103,17 +103,25 @@ class DialQueueManager {
    * @param {DialRequest} dialRequest
    * @returns {void}
    */
-  add ({ peerInfo, protocol, useFSM, callback }) {
+  add ({ peerInfo, protocol, options, callback }) {
     callback = callback ? once(callback) : noop
 
     // Add the dial to its respective queue
     const targetQueue = this.getQueue(peerInfo)
-    // If we have too many cold calls, abort the dial immediately
-    if (this._coldCallQueue.size >= this.switch.dialer.MAX_COLD_CALLS && !protocol) {
-      return nextTick(callback, DIAL_ABORTED())
+
+    // Cold Call
+    if (options.priority > PRIORITY_HIGH) {
+      // If we have too many cold calls, abort the dial immediately
+      if (this._coldCallQueue.size >= this.switch.dialer.MAX_COLD_CALLS) {
+        return nextTick(callback, DIAL_ABORTED())
+      }
+
+      if (this._queue.has(targetQueue.id)) {
+        return nextTick(callback, DIAL_ABORTED())
+      }
     }
 
-    targetQueue.add(protocol, useFSM, callback)
+    targetQueue.add(protocol, options.useFSM, callback)
 
     // If we're already connected to the peer, start the queue now
     // While it might cause queues to go over the max parallel amount,
@@ -130,15 +138,12 @@ class DialQueueManager {
 
     // Add the id to its respective queue set if the queue isn't running
     if (!targetQueue.isRunning) {
-      if (protocol) {
+      if (options.priority <= PRIORITY_HIGH) {
         this._queue.add(targetQueue.id)
         this._coldCallQueue.delete(targetQueue.id)
       // Only add it to the cold queue if it's not in the normal queue
-      } else if (!this._queue.has(targetQueue.id)) {
-        this._coldCallQueue.add(targetQueue.id)
-      // The peer is already in the normal queue, abort the cold call
       } else {
-        return nextTick(callback, DIAL_ABORTED())
+        this._coldCallQueue.add(targetQueue.id)
       }
     }
 

test/dialer.spec.js

@@ -12,17 +12,20 @@ const PeerBook = require('peer-book')
 const Queue = require('../src/dialer/queue')
 const QueueManager = require('../src/dialer/queueManager')
 const Switch = require('../src')
+const { PRIORITY_HIGH, PRIORITY_LOW } = require('../src/constants')
 
 const utils = require('./utils')
 const createInfos = utils.createInfos
 
 describe('dialer', () => {
   let switchA
+  let switchB
 
-  before((done) => createInfos(1, (err, infos) => {
+  before((done) => createInfos(2, (err, infos) => {
     expect(err).to.not.exist()
 
     switchA = new Switch(infos[0], new PeerBook())
+    switchB = new Switch(infos[1], new PeerBook())
 
     done()
   }))
@@ -31,6 +34,26 @@ describe('dialer', () => {
     sinon.restore()
   })
 
+  describe('connect', () => {
+    afterEach(() => {
+      switchA.dialer.clearBlacklist(switchB._peerInfo)
+    })
+
+    it('should use default options', (done) => {
+      switchA.dialer.connect(switchB._peerInfo, (err) => {
+        expect(err).to.exist()
+        done()
+      })
+    })
+
+    it('should be able to use custom options', (done) => {
+      switchA.dialer.connect(switchB._peerInfo, { useFSM: true, priority: PRIORITY_HIGH }, (err) => {
+        expect(err).to.exist()
+        done()
+      })
+    })
+  })
+
   describe('queue', () => {
     it('should blacklist forever after 5 blacklists', () => {
       const queue = new Queue('QM', switchA)
@@ -58,7 +81,7 @@ describe('dialer', () => {
           id: { toB58String: () => 'QmA' }
         },
         protocol: null,
-        useFSM: true,
+        options: { useFSM: true, priority: PRIORITY_LOW },
         callback: (err) => {
           expect(err.code).to.eql('DIAL_ABORTED')
           done()
@@ -75,7 +98,7 @@ describe('dialer', () => {
           isConnected: () => null
         },
         protocol: '/echo/1.0.0',
-        useFSM: true,
+        options: { useFSM: true, priority: PRIORITY_HIGH },
         callback: () => {}
       }
 
@@ -99,7 +122,7 @@ describe('dialer', () => {
           isConnected: () => null
         },
         protocol: null,
-        useFSM: true,
+        options: { useFSM: true, priority: PRIORITY_LOW },
         callback: () => {}
       }
 
@@ -120,7 +143,7 @@ describe('dialer', () => {
           isConnected: () => null
         },
         protocol: null,
-        useFSM: true,
+        options: { useFSM: true, priority: PRIORITY_LOW },
         callback: (err) => {
           expect(runSpy.called).to.eql(false)
           expect(hasSpy.called).to.eql(true)