feat(transport): use parallel limited dialer

Replaces #193
Fixes #194

Author: dignifiedquire

package.json

@@ -86,4 +86,4 @@
     "greenkeeper[bot] <greenkeeper[bot]@users.noreply.github.com>",
     "ᴠɪᴄᴛᴏʀ ʙᴊᴇʟᴋʜᴏʟᴍ <[email protected]>"
   ]
-}
+}

src/transport.js

@@ -2,13 +2,19 @@
 
 const Connection = require('interface-connection').Connection
 const parallel = require('async/parallel')
+const queue = require('async/queue')
 const once = require('once')
 const debug = require('debug')
 const log = debug('libp2p:swarm:transport')
 
 const protocolMuxer = require('./protocol-muxer')
 
+// number of concurrent outbound dials to make per peer, same as go-libp2p-swarm
+const defaultPerPeerRateLimit = 8
+
 module.exports = function (swarm) {
+  const queues = new Map()
+
   return {
     add (key, transport, options, callback) {
       if (typeof options === 'function') {
@@ -36,32 +42,76 @@ module.exports = function (swarm) {
         multiaddrs = [multiaddrs]
       }
       log('dialing %s', key, multiaddrs.map((m) => m.toString()))
-      // a) filter the multiaddrs that are actually valid for this transport (use a func from the transport itself) (maybe even make the transport do that)
+      // filter the multiaddrs that are actually valid for this transport (use a func from the transport itself) (maybe even make the transport do that)
       multiaddrs = dialables(t, multiaddrs)
 
-      // b) if multiaddrs.length = 1, return the conn from the
-      // transport, otherwise, create a passthrough
-      if (multiaddrs.length === 1) {
-        const conn = t.dial(multiaddrs.shift(), (err) => {
-          if (err) return callback(err)
-          callback(null, new Connection(conn))
-        })
-        return
-      }
+      // create dial queue if non exists
+      let q
+      if (queues.has(key)) {
+        log('reusing queue')
+        q = queues.get(key)
+      } else {
+        log('setting up new queue')
+        q = queue((multiaddr, cb) => {
+          const conn = t.dial(multiaddr, (err) => {
+            if (err) {
+              log('dial failed: %s', multiaddr.toString())
+              return cb(err)
+            }
+            if (q.canceled) {
+              log('dial canceled: %s', multiaddr.toString())
+              // clean up already done dials
+
+              return cb()
+            }
+            // one is enough
+            log('dial success: %s', multiaddr.toString())
+            q.kill()
+            q.canceled = true
+
+            q.finish(null, conn)
+          })
+        }, defaultPerPeerRateLimit)
+
+        q.errors = []
+        q.finishCbs = []
 
-      // c) multiaddrs should already be a filtered list
-      // specific for the transport we are using
-      const proxyConn = new Connection()
+        // handle finish
+        q.finish = (err, conn) => {
+          log('queue finish')
+          queues.delete(key)
 
-      next(multiaddrs.shift())
+          q.finishCbs.forEach((next) => {
+            if (err) {
+              next(err)
+            } else {
+              const proxyConn = new Connection()
+              proxyConn.setInnerConn(conn)
+
+              next(null, proxyConn)
+            }
+          })
+        }
 
-      // TODO improve in the future to make all the dials in paralell
-      function next (multiaddr) {
-        const conn = t.dial(multiaddr, () => {
-          proxyConn.setInnerConn(conn)
-          callback(null, proxyConn)
-        })
+        // collect errors
+        q.error = (err) => {
+          q.errors.push(err)
+        }
+
+        // no more addresses and all failed
+        q.drain = () => {
+          log('queue drain')
+          const err = new Error('Could not dial any address')
+          err.errors = q.errors
+          q.errors = []
+          q.finish(err)
+        }
+
+        queues.set(key, q)
       }
+
+      q.push(multiaddrs)
+      q.finishCbs.push(callback)
     },
 
     listen (key, options, handler, callback) {