Abort requests when network downgrading

When the network becomes slow, we check if stopping the current request
and download the content with lower resolution is faster. If so, abort
the current request and start a new one.

Issue #1051

Change-Id: I588e524469432e362361d1cfbde6cd45c2009959

externs/shaka/net.js

@@ -145,7 +145,7 @@ shaka.extern.SchemePlugin;
 
 
 /**
- * @typedef {function(number, number)}
+ * @typedef {function(number, number, number)}
  *
  * @description
  * A callback function to handle progress event through networking engine in
@@ -154,7 +154,8 @@ shaka.extern.SchemePlugin;
  * took to complete.
  * The second argument is the total number of bytes downloaded during that
  * time.
- *
+ * The third argument is the number of bytes remaining to be loaded in a
+ * segment.
  * @exportDoc
  */
 shaka.extern.ProgressUpdated;

lib/media/segment_reference.js

@@ -209,6 +209,19 @@ shaka.media.SegmentReference.prototype.getEndByte = function() {
 };
 
 
+/**
+ * Returns the size of the segment.
+ * @return {?number}
+ */
+shaka.media.SegmentReference.prototype.getSize = function() {
+  if (this.endByte) {
+    return this.endByte - this.startByte;
+  } else {
+    return null;
+  }
+};
+
+
 /**
  * A convenient typedef for when either type of reference is acceptable.
  *

lib/media/streaming_engine.js

@@ -242,7 +242,8 @@ shaka.media.StreamingEngine.PlayerInterface;
  *   clearingBuffer: boolean,
  *   recovering: boolean,
  *   hasError: boolean,
- *   resumeAt: number
+ *   resumeAt: number,
+ *   operation: shaka.net.NetworkingEngine.PendingRequest
  * }}
  *
  * @description
@@ -291,6 +292,8 @@ shaka.media.StreamingEngine.PlayerInterface;
  *   An override for the time to start performing updates at.  If the playhead
  *   is behind this time, update_() will still start fetching segments from
  *   this time.  If the playhead is ahead of the time, this field is ignored.
+ * @property {shaka.net.NetworkingEngine.PendingRequest} operation
+ *   Operation with the number of bytes to be downloaded.
  */
 shaka.media.StreamingEngine.MediaState_;
 
@@ -693,7 +696,7 @@ shaka.media.StreamingEngine.prototype.switchInternal_ = function(
   if (!mediaState) return;
 
   // If we are selecting a stream from a different Period, then we need to
-  // handle a Period transition.  Simply ignore the given stream, assuming that
+  // handle a Period transition. Simply ignore the given stream, assuming that
   // Player will select the same track in onChooseStreams.
   let periodIndex = this.findPeriodContainingStream_(stream);
   if (clearBuffer && periodIndex != mediaState.needPeriodIndex) {
@@ -755,6 +758,11 @@ shaka.media.StreamingEngine.prototype.switchInternal_ = function(
   let streamTag = shaka.media.StreamingEngine.logPrefix_(mediaState);
   shaka.log.debug('switch: switching to Stream ' + streamTag);
 
+  if (this.shouldAbortCurrentRequest_(mediaState, periodIndex)) {
+    shaka.log.info('Aborting current segment request to switch.');
+    mediaState.operation.abort();
+  }
+
   if (clearBuffer) {
     if (mediaState.clearingBuffer) {
       // We are already going to clear the buffer, but make sure it is also
@@ -778,6 +786,52 @@ shaka.media.StreamingEngine.prototype.switchInternal_ = function(
 };
 
 
+/**
+ * Return true if we should abort the current request.
+ * Compare the current bytes remaining and the stream with lower
+ * resolution's number of bytes.
+ * If the network is slow, and the new stream's size is smaller than the current
+ * bytes remaining, abort the current request and start to download the new one.
+ * TODO: if we have downloaded most of the segment, ex. 95%, we should not abort
+ * the current request.
+ *
+ * @param {!shaka.media.StreamingEngine.MediaState_} mediaState
+ * @param {number} periodIndex
+ * @return {boolean}
+ */
+shaka.media.StreamingEngine.prototype.shouldAbortCurrentRequest_ =
+    function(mediaState, periodIndex) {
+  // If the operation is completed, it will be set to null, and there's no need
+  // to abort the request.
+  if (!mediaState.operation) {
+    return false;
+  }
+
+  const presentationTime = this.playerInterface_.getPresentationTime();
+  const bufferEnd =
+      this.playerInterface_.mediaSourceEngine.bufferEnd(mediaState.type);
+
+  // The new segment with lower resolution that we're about to switch to.
+  const lowerResSegment = this.getSegmentReferenceNeeded_(
+      mediaState, presentationTime, bufferEnd, periodIndex);
+  const lowerResSegmentSize = lowerResSegment ? lowerResSegment.getSize(): -1;
+
+  const lowerResInitSegmentRef = mediaState.stream.initSegmentReference;
+  // If we don't know the size of the init segment of the new stream, we can't
+  // decide to abort the current request.
+  if (!lowerResInitSegmentRef || !lowerResInitSegmentRef.endByte) {
+    return false;
+  }
+  const lowerResInitSegmentSize =
+      lowerResInitSegmentRef.endByte - lowerResInitSegmentRef.startByte;
+
+  const bytesRemaining = mediaState.operation.getBytesRemaining();
+
+  return lowerResSegmentSize > 0 &&
+      (bytesRemaining > lowerResSegmentSize + lowerResInitSegmentSize);
+};
+
+
 /**
  * Notifies the StreamingEngine that the playhead has moved to a valid time
  * within the presentation timeline.
@@ -960,7 +1014,8 @@ shaka.media.StreamingEngine.prototype.createMediaState_ = function(
     clearingBuffer: false,
     recovering: false,
     hasError: false,
-    resumeAt: resumeAt,
+    resumeAt: resumeAt || 0,
+    operation: null,
   });
 };
 
@@ -1562,7 +1617,8 @@ shaka.media.StreamingEngine.prototype.fetchAndAppend_ = function(
   mediaState.needInitSegment = false;
 
   shaka.log.v2(logPrefix, 'fetching segment');
-  let fetchSegment = this.fetch_(reference);
+  let fetchSegment = this.fetch_(mediaState, reference);
+
 
   Promise.all([initSourceBuffer, fetchSegment]).then(function(results) {
     if (this.destroyed_ || this.fatalError_) return;
@@ -1607,8 +1663,13 @@ shaka.media.StreamingEngine.prototype.fetchAndAppend_ = function(
         shaka.log.warning(logPrefix,
             'Text stream failed to parse. Proceeding without it.');
       }
-
       this.mediaStates_.delete(ContentType.TEXT);
+    } else if (error.code == shaka.util.Error.Code.OPERATION_ABORTED) {
+      // If the network slows down, abort the current fetch request and start a
+      // new one, and ignore the error message.
+      mediaState.performingUpdate = false;
+      mediaState.updateTimer = null;
+      this.scheduleUpdate_(mediaState, 0);
     } else if (error.code == shaka.util.Error.Code.QUOTA_EXCEEDED_ERROR) {
       this.handleQuotaExceeded_(mediaState, error);
     } else {
@@ -1760,7 +1821,9 @@ shaka.media.StreamingEngine.prototype.initSourceBuffer_ = function(
   }
 
   shaka.log.v1(logPrefix, 'fetching init segment');
-  let fetchInit = this.fetch_(mediaState.stream.initSegmentReference);
+
+  let fetchInit =
+      this.fetch_(mediaState, mediaState.stream.initSegmentReference);
   let appendInit = fetchInit.then(function(initSegment) {
     if (this.destroyed_) return;
     shaka.log.v1(logPrefix, 'appending init segment');
@@ -2253,13 +2316,14 @@ shaka.media.StreamingEngine.prototype.findPeriodContainingStream_ = function(
 /**
  * Fetches the given segment.
  *
+ * @param {!shaka.media.StreamingEngine.MediaState_} mediaState
  * @param {(!shaka.media.InitSegmentReference|!shaka.media.SegmentReference)}
  *   reference
  *
  * @return {!Promise.<!ArrayBuffer>}
  * @private
  */
-shaka.media.StreamingEngine.prototype.fetch_ = function(reference) {
+shaka.media.StreamingEngine.prototype.fetch_ = function(mediaState, reference) {
   const requestType = shaka.net.NetworkingEngine.RequestType.SEGMENT;
 
   const request = shaka.util.Networking.createSegmentRequest(
@@ -2270,8 +2334,10 @@ shaka.media.StreamingEngine.prototype.fetch_ = function(reference) {
 
   shaka.log.v2('fetching: reference=', reference);
 
-  let op = this.playerInterface_.netEngine.request(requestType, request);
+  const op = this.playerInterface_.netEngine.request(requestType, request);
+  mediaState.operation = op;
   return op.promise.then(function(response) {
+    mediaState.operation = null;
     return response.data;
   });
 };

lib/net/http_fetch_plugin.js

@@ -32,7 +32,7 @@ goog.require('shaka.util.Timer');
  * @param {string} uri
  * @param {shaka.extern.Request} request
  * @param {shaka.net.NetworkingEngine.RequestType} requestType
- * @param {function(number, number)=} progressUpdated Called when a progress
+ * @param {shaka.extern.ProgressUpdated=} progressUpdated Called when a progress
  *        event happened.
  * @return {!shaka.extern.IAbortableOperation.<shaka.extern.Response>}
  * @export
@@ -98,7 +98,7 @@ shaka.net.HttpFetchPlugin = function(
  * @param {shaka.net.NetworkingEngine.RequestType} requestType
  * @param {!RequestInit} init
  * @param {shaka.net.HttpFetchPlugin.AbortStatus} abortStatus
- * @param {function(number, number)=} progressUpdated
+ * @param {shaka.extern.ProgressUpdated=} progressUpdated
  * @return {!Promise<!shaka.extern.Response>}
  * @private
  */
@@ -126,6 +126,10 @@ shaka.net.HttpFetchPlugin.request_ = async function(
     // stream; if we didn't clone it here, we would be unable to get the
     // response's arrayBuffer later.
     const reader = response.clone().body.getReader();
+
+    const contentLengthRaw = response.headers.get('Content-Length');
+    const contentLength = contentLengthRaw ? parseInt(contentLengthRaw, 10) : 0;
+
     let start = (controller) => {
       let push = async () => {
         const readObj = await reader.read();
@@ -139,7 +143,8 @@ shaka.net.HttpFetchPlugin.request_ = async function(
         // is long enough, or if a whole segment is downloaded, call
         // progressUpdated().
         if (currentTime - lastTime > 100 || readObj.done) {
-          progressUpdated(currentTime - lastTime, loaded - lastLoaded);
+          progressUpdated(currentTime - lastTime, loaded - lastLoaded,
+              contentLength - loaded);
           lastLoaded = loaded;
           lastTime = currentTime;
         }

lib/net/http_xhr_plugin.js

@@ -30,8 +30,8 @@ goog.require('shaka.util.Error');
  * @param {string} uri
  * @param {shaka.extern.Request} request
  * @param {shaka.net.NetworkingEngine.RequestType} requestType
- * @param {function(number, number)=} progressUpdated Called when a progress
- *        event happened.
+ * @param {shaka.extern.ProgressUpdated=} progressUpdated Called when a
+ *        progress event happened.
  * @return {!shaka.extern.IAbortableOperation.<shaka.extern.Response>}
  * @export
  */
@@ -101,7 +101,8 @@ shaka.net.HttpXHRPlugin = function(uri, request, requestType, progressUpdated) {
       // progressUpdated().
       if (currentTime - lastTime > 100 ||
           (event.lengthComputable && event.loaded == event.total)) {
-        progressUpdated(currentTime - lastTime, event.loaded - lastLoaded);
+        progressUpdated(currentTime - lastTime, event.loaded - lastLoaded,
+          event.total - event.loaded);
         lastLoaded = event.loaded;
         lastTime = currentTime;
       }