Merge pull request #1923 from webtorrent/add_utp_support

uTP support (disabled by default). when enabled, uTP is tried first for all ip:port and timeouts fall back to TCP

Co-authored-by: Pierre Dubouilh <pldubouilh@gmail.com>

3 files changed, 129 insertions, 44 deletions

diff --git a/lib/tcp-pool.js b/lib/conn-pool.js
index 1094c4f..5746ca5 100644
--- a/lib/tcp-pool.js
+++ b/lib/conn-pool.js
@@ -1,73 +1,112 @@
 const arrayRemove = require('unordered-array-remove')
-const debug = require('debug')('webtorrent:tcp-pool')
+const debug = require('debug')('webtorrent:conn-pool')
 const net = require('net') // browser exclude
+const utp = require('utp-native') // browser exclude
 
 const Peer = require('./peer')
 
 /**
- * TCPPool
+ * Connection Pool
  *
- * A "TCP pool" allows multiple swarms to listen on the same TCP port and determines
+ * A connection pool allows multiple swarms to listen on the same TCP/UDP port and determines
  * which swarm incoming connections are intended for by inspecting the bittorrent
  * handshake that the remote peer sends.
  *
  * @param {number} port
  */
-class TCPPool {
+class ConnPool {
   constructor (client) {
-    debug('create tcp pool (port %s)', client.torrentPort)
+    debug('create pool (port %s)', client.torrentPort)
 
-    this.server = net.createServer()
     this._client = client
 
     // Temporarily store incoming connections so they can be destroyed if the server is
     // closed before the connection is passed off to a Torrent.
     this._pendingConns = []
 
-    this._onConnectionBound = conn => {
-      this._onConnection(conn)
+    this._onTCPConnectionBound = (conn) => {
+      this._onConnection(conn, 'tcp')
+    }
+
+    this._onUTPConnectionBound = (conn) => {
+      this._onConnection(conn, 'utp')
     }
 
     this._onListening = () => {
       this._client._onListening()
     }
 
-    this._onError = err => {
+    this._onTCPError = (err) => {
       this._client._destroy(err)
     }
 
-    this.server.on('connection', this._onConnectionBound)
-    this.server.on('listening', this._onListening)
-    this.server.on('error', this._onError)
+    this._onUTPError = () => {
+      this._client.utp = false
+    }
 
-    this.server.listen(client.torrentPort)
+    // Setup TCP
+    this.tcpServer = net.createServer()
+    this.tcpServer.on('connection', this._onTCPConnectionBound)
+    this.tcpServer.on('error', this._onTCPError)
+
+    // Start TCP
+    this.tcpServer.listen(client.torrentPort, () => {
+      debug('creating tcpServer in port %s', this.tcpServer.address().port)
+      if (this._client.utp) {
+        // Setup uTP
+        this.utpServer = utp.createServer()
+        this.utpServer.on('connection', this._onUTPConnectionBound)
+        this.utpServer.on('listening', this._onListening)
+        this.utpServer.on('error', this._onUTPError)
+
+        // Start uTP
+        debug('creating utpServer in port %s', this.tcpServer.address().port)
+        this.utpServer.listen(this.tcpServer.address().port)
+      } else {
+        this._onListening()
+      }
+    })
   }
 
   /**
-   * Destroy this TCP pool.
+   * Destroy this Conn pool.
    * @param  {function} cb
    */
   destroy (cb) {
-    debug('destroy tcp pool')
+    debug('destroy conn pool')
 
-    this.server.removeListener('connection', this._onConnectionBound)
-    this.server.removeListener('listening', this._onListening)
-    this.server.removeListener('error', this._onError)
+    if (this.utpServer) {
+      this.utpServer.removeListener('connection', this._onUTPConnectionBound)
+      this.utpServer.removeListener('listening', this._onListening)
+      this.utpServer.removeListener('error', this._onUTPError)
+    }
+
+    this.tcpServer.removeListener('connection', this._onTCPConnectionBound)
+    this.tcpServer.removeListener('error', this._onTCPError)
 
     // Destroy all open connection objects so server can close gracefully without waiting
     // for connection timeout or remote peer to disconnect.
-    this._pendingConns.forEach(conn => {
+    this._pendingConns.forEach((conn) => {
       conn.on('error', noop)
       conn.destroy()
     })
 
+    if (this.utpServer) {
+      try {
+        this.utpServer.close(cb)
+      } catch (err) {
+        if (cb) process.nextTick(cb)
+      }
+    }
+
     try {
-      this.server.close(cb)
+      this.tcpServer.close(cb)
     } catch (err) {
       if (cb) process.nextTick(cb)
     }
 
-    this.server = null
+    this.tcpServer = null
+    this.utpServer = null
     this._client = null
     this._pendingConns = null
   }
@@ -76,7 +115,7 @@ class TCPPool {
    * On incoming connections, we expect the remote peer to send a handshake first. Based
    * on the infoHash in that handshake, route the peer to the right swarm.
    */
-  _onConnection (conn) {
+  _onConnection (conn, type) {
     const self = this
 
     // If the connection has already been closed before the `connect` event is fired,
@@ -92,7 +131,7 @@ class TCPPool {
     self._pendingConns.push(conn)
     conn.once('close', cleanupPending)
 
-    const peer = Peer.createTCPIncomingPeer(conn)
+    const peer = type === 'utp' ? Peer.createUTPIncomingPeer(conn) : Peer.createTCPIncomingPeer(conn)
 
     const wire = peer.wire
     wire.once('handshake', onHandshake)
@@ -125,4 +164,4 @@ class TCPPool {
 
 function noop () {}
 
-module.exports = TCPPool
+module.exports = ConnPool
diff --git a/lib/peer.js b/lib/peer.js
index 0ddf535..0d0b807 100644
--- a/lib/peer.js
+++ b/lib/peer.js
@@ -5,6 +5,7 @@ const Wire = require('bittorrent-protocol')
 const WebConn = require('./webconn')
 
 const CONNECT_TIMEOUT_TCP = 5000
+const CONNECT_TIMEOUT_UTP = 5000
 const CONNECT_TIMEOUT_WEBRTC = 25000
 const HANDSHAKE_TIMEOUT = 25000
 
@@ -35,8 +36,37 @@ exports.createWebRTCPeer = (conn, swarm) => {
  * know what swarm the connection is intended for.
  */
 exports.createTCPIncomingPeer = conn => {
+  return _createIncomingPeer(conn, 'tcpIncoming')
+}
+
+/**
+ * Incoming uTP peers start out connected, because the remote peer connected to the
+ * listening port of the uTP server. Until the remote peer sends a handshake, we don't
+ * know what swarm the connection is intended for.
+ */
+exports.createUTPIncomingPeer = conn => {
+  return _createIncomingPeer(conn, 'utpIncoming')
+}
+
+/**
+ * Outgoing TCP peers start out with just an IP address. At some point (when there is an
+ * available connection), the client can attempt to connect to the address.
+ */
+exports.createTCPOutgoingPeer = (addr, swarm) => {
+  return _createOutgoingPeer(addr, swarm, 'tcpOutgoing')
+}
+
+/**
+ * Outgoing uTP peers start out with just an IP address. At some point (when there is an
+ * available connection), the client can attempt to connect to the address.
+ */
+exports.createUTPOutgoingPeer = (addr, swarm) => {
+  return _createOutgoingPeer(addr, swarm, 'utpOutgoing')
+}
+
+const _createIncomingPeer = (conn, type) => {
   const addr = `${conn.remoteAddress}:${conn.remotePort}`
-  const peer = new Peer(addr, 'tcpIncoming')
+  const peer = new Peer(addr, type)
   peer.conn = conn
   peer.addr = addr
 
@@ -45,12 +75,8 @@ exports.createTCPIncomingPeer = conn => {
   return peer
 }
 
-/**
- * Outgoing TCP peers start out with just an IP address. At some point (when there is an
- * available connection), the client can attempt to connect to the address.
- */
-exports.createTCPOutgoingPeer = (addr, swarm) => {
-  const peer = new Peer(addr, 'tcpOutgoing')
+const _createOutgoingPeer = (addr, swarm, type) => {
+  const peer = new Peer(addr, type)
   peer.addr = addr
   peer.swarm = swarm
 
@@ -193,9 +219,16 @@ class Peer {
 
   startConnectTimeout () {
     clearTimeout(this.connectTimeout)
+
+    const connectTimeoutValues = {
+      webrtc: CONNECT_TIMEOUT_WEBRTC,
+      tcpOutgoing: CONNECT_TIMEOUT_TCP,
+      utpOutgoing: CONNECT_TIMEOUT_UTP
+    }
+
     this.connectTimeout = setTimeout(() => {
       this.destroy(new Error('connect timeout'))
-    }, this.type === 'webrtc' ? CONNECT_TIMEOUT_WEBRTC : CONNECT_TIMEOUT_TCP)
+    }, connectTimeoutValues[this.type])
     if (this.connectTimeout.unref) this.connectTimeout.unref()
   }
 
@@ -212,7 +245,7 @@ class Peer {
     this.destroyed = true
     this.connected = false
 
-    debug('destroy %s (error: %s)', this.id, err && (err.message || err))
+    debug('destroy %s %s (error: %s)', this.type, this.id, err && (err.message || err))
 
     clearTimeout(this.connectTimeout)
     clearTimeout(this.handshakeTimeout)
diff --git a/lib/torrent.js b/lib/torrent.js
index e2ee673..f24ccf3 100644
--- a/lib/torrent.js
+++ b/lib/torrent.js
@@ -24,6 +24,7 @@ const sha1 = require('simple-sha1')
 const speedometer = require('speedometer')
 const utMetadata = require('ut_metadata')
 const utPex = require('ut_pex') // browser exclude
+const utp = require('utp-native') // browser exclude
 const parseRange = require('parse-numeric-range')
 
 const File = require('./file')
@@ -750,7 +751,8 @@ class Torrent extends EventEmitter {
       }
     }
 
-    const wasAdded = !!this._addPeer(peer)
+    // if the utp connection fails to connect, then it is replaced with a tcp connection to the same ip:port
+    const wasAdded = !!this._addPeer(peer, this.client.utp ? 'utp' : 'tcp')
     if (wasAdded) {
       this.emit('peer', peer)
     } else {
@@ -759,7 +761,7 @@ class Torrent extends EventEmitter {
     return wasAdded
   }
 
-  _addPeer (peer) {
+  _addPeer (peer, type) {
     if (this.destroyed) {
       if (typeof peer !== 'string') peer.destroy()
       return null
@@ -787,7 +789,7 @@ class Torrent extends EventEmitter {
     let newPeer
     if (typeof peer === 'string') {
       // `peer` is an addr ("ip:port" string)
-      newPeer = Peer.createTCPOutgoingPeer(peer, this)
+      newPeer = type === 'utp' ? Peer.createUTPOutgoingPeer(peer, this) : Peer.createTCPOutgoingPeer(peer, this)
     } else {
       // `peer` is a WebRTC connection (simple-peer)
       newPeer = Peer.createWebRTCPeer(peer, this)
@@ -1682,7 +1684,7 @@ class Torrent extends EventEmitter {
     const peer = this._queue.shift()
     if (!peer) return // queue could be empty
 
-    this._debug('tcp connect attempt to %s', peer.addr)
+    this._debug('%s connect attempt to %s', peer.type, peer.addr)
 
     const parts = addrToIPPort(peer.addr)
     const opts = {
@@ -1690,7 +1692,13 @@ class Torrent extends EventEmitter {
       port: parts[1]
     }
 
-    const conn = peer.conn = net.connect(opts)
+    if (peer.type === 'utpOutgoing') {
+      peer.conn = utp.connect(opts.port, opts.host)
+    } else {
+      peer.conn = net.connect(opts)
+    }
+
+    const conn = peer.conn
 
     conn.once('connect', () => { peer.onConnect() })
     conn.once('error', err => { peer.destroy(err) })
@@ -1703,11 +1711,16 @@ class Torrent extends EventEmitter {
       // TODO: If torrent is done, do not try to reconnect after a timeout
 
       if (peer.retries >= RECONNECT_WAIT.length) {
-        this._debug(
-          'conn %s closed: will not re-add (max %s attempts)',
-          peer.addr, RECONNECT_WAIT.length
-        )
-        return
+        if (this.client.utp) {
+          const newPeer = this._addPeer(peer.addr, 'tcp')
+          if (newPeer) newPeer.retries = 0
+        } else {
+          this._debug(
+            'conn %s closed: will not re-add (max %s attempts)',
+            peer.addr, RECONNECT_WAIT.length
+          )
+          return
+        }
       }
 
       const ms = RECONNECT_WAIT[peer.retries]
@@ -1717,7 +1730,7 @@ class Torrent extends EventEmitter {
       )
 
       const reconnectTimeout = setTimeout(() => {
-        const newPeer = this._addPeer(peer.addr)
+        const newPeer = this._addPeer(peer.addr, this.client.utp ? 'utp' : 'tcp')
         if (newPeer) newPeer.retries = peer.retries + 1
       }, ms)
       if (reconnectTimeout.unref) reconnectTimeout.unref()