dynamic requests pipeline length

lib/torrent.js

@@ -21,11 +21,13 @@ var ut_metadata = require('ut_metadata')
 var ut_pex = require('ut_pex') // browser exclude
 
 var MAX_BLOCK_LENGTH = 128 * 1024
-var MAX_OUTSTANDING_REQUESTS = 5
 var PIECE_TIMEOUT = 10000
 var CHOKE_TIMEOUT = 5000
 var SPEED_THRESHOLD = 3 * Storage.BLOCK_LENGTH
 
+var PIPELINE_MIN_DURATION = 0.5
+var PIPELINE_MAX_DURATION = 1
+
 var RECHOKE_INTERVAL = 10000 // 10 seconds
 var RECHOKE_OPTIMISTIC_DURATION = 2 // 30 seconds
 
@@ -650,6 +652,10 @@ Torrent.prototype._updateWire = function (wire) {
   if (wire.peerChoking) return
   if (!wire.downloaded) return validateWire()
 
+  var minOutstandingRequests = getPipelineLength(wire, PIPELINE_MIN_DURATION)
+  if (wire.requests.length >= minOutstandingRequests) return true
+  var maxOutstandingRequests = getPipelineLength(wire, PIPELINE_MAX_DURATION)
+
   trySelectWire(false) || trySelectWire(true)
 
   function genPieceFilterFunc (start, end, tried, rank) {
@@ -696,7 +702,7 @@ Torrent.prototype._updateWire = function (wire) {
     var speed = wire.downloadSpeed() || 1
     if (speed > SPEED_THRESHOLD) return function () { return true }
 
-    var secs = MAX_OUTSTANDING_REQUESTS * Storage.BLOCK_LENGTH / speed
+    var secs = Math.max(1, wire.requests.length) * Storage.BLOCK_LENGTH / speed
     var tries = 10
     var ptr = 0
 
@@ -734,7 +740,7 @@ Torrent.prototype._updateWire = function (wire) {
   }
 
   function trySelectWire (hotswap) {
-    if (wire.requests.length >= MAX_OUTSTANDING_REQUESTS) return true
+    if (wire.requests.length >= maxOutstandingRequests) return true
     var rank = speedRanker()
 
     for (var i = 0; i < self._selections.length; i++) {
@@ -756,7 +762,7 @@ Torrent.prototype._updateWire = function (wire) {
           // request all non-reserved blocks in this piece
           while (self._request(wire, piece, self._critical[piece] || hotswap)) {}
 
-          if (wire.requests.length < MAX_OUTSTANDING_REQUESTS) {
+          if (wire.requests.length < maxOutstandingRequests) {
             tried[piece] = true
             tries++
             continue
@@ -772,7 +778,7 @@ Torrent.prototype._updateWire = function (wire) {
           // request all non-reserved blocks in piece
           while (self._request(wire, piece, self._critical[piece] || hotswap)) {}
 
-          if (wire.requests.length < MAX_OUTSTANDING_REQUESTS) continue
+          if (wire.requests.length < maxOutstandingRequests) continue
 
           if (next.priority) shufflePriority(i)
           return true
@@ -915,7 +921,8 @@ Torrent.prototype._request = function (wire, index, hotswap) {
   var numRequests = wire.requests.length
 
   if (self.storage.bitfield.get(index)) return false
-  if (numRequests >= MAX_OUTSTANDING_REQUESTS) return false
+  var maxOutstandingRequests = getPipelineLength(wire, PIPELINE_MAX_DURATION)
+  if (numRequests >= maxOutstandingRequests) return false
 
   var endGame = (wire.requests.length === 0 && self.storage.numMissing < 30)
   var block = self.storage.reserveBlock(index, endGame)
@@ -972,6 +979,10 @@ Torrent.prototype.createServer = function (opts) {
   }
 }
 
+function getPipelineLength (wire, duration) {
+  return Math.ceil(2 + duration * wire.downloadSpeed() / Storage.BLOCK_LENGTH);
+}
+
 /**
  * Returns a random integer in [0,high)
  */