Bug 1175768 - Throttle NotifyDataArrived. r=jya

dom/media/AbstractMediaDecoder.h

@@ -117,7 +117,8 @@ public:
 
   // Called by the reader's MediaResource as data arrives over the network.
   // Must be called on the main thread.
-  virtual void NotifyDataArrived(uint32_t aLength, int64_t aOffset) = 0;
+  virtual void NotifyDataArrived(uint32_t aLength, int64_t aOffset,
+                                 bool aThrottleUpdates) = 0;
 
   // Set by Reader if the current audio track can be offloaded
   virtual void SetPlatformCanOffloadAudio(bool aCanOffloadAudio) {}

dom/media/MediaDecoder.cpp

@@ -1313,11 +1313,11 @@ size_t MediaDecoder::SizeOfAudioQueue() {
   return 0;
 }
 
-void MediaDecoder::NotifyDataArrived(uint32_t aLength, int64_t aOffset) {
+void MediaDecoder::NotifyDataArrived(uint32_t aLength, int64_t aOffset, bool aThrottleUpdates) {
   MOZ_ASSERT(NS_IsMainThread());
 
   if (mDecoderStateMachine) {
-    mDecoderStateMachine->DispatchNotifyDataArrived(aLength, aOffset);
+    mDecoderStateMachine->DispatchNotifyDataArrived(aLength, aOffset, aThrottleUpdates);
   }
 
   // ReadyState computation depends on MediaDecoder::CanPlayThrough, which

dom/media/MediaDecoder.h

@@ -434,7 +434,8 @@ public:
 
   // Called as data arrives on the stream and is read into the cache.  Called
   // on the main thread only.
-  virtual void NotifyDataArrived(uint32_t aLength, int64_t aOffset) override;
+  virtual void NotifyDataArrived(uint32_t aLength, int64_t aOffset,
+                                 bool aThrottleUpdates) override;
 
   // Called by MediaResource when the principal of the resource has
   // changed. Called on main thread only.

dom/media/MediaDecoderReader.cpp

@@ -70,8 +70,11 @@ MediaDecoderReader::MediaDecoderReader(AbstractMediaDecoder* aDecoder,
                                   : new MediaTaskQueue(GetMediaThreadPool(MediaThreadType::PLAYBACK),
                                                        /* aSupportsTailDispatch = */ true))
   , mWatchManager(this, mTaskQueue)
+  , mTimer(new MediaTimer())
   , mBuffered(mTaskQueue, TimeIntervals(), "MediaDecoderReader::mBuffered (Canonical)")
   , mDuration(mTaskQueue, NullableTimeUnit(), "MediaDecoderReader::mDuration (Mirror)")
+  , mThrottleDuration(TimeDuration::FromMilliseconds(500))
+  , mLastThrottledNotify(TimeStamp::Now() - mThrottleDuration)
   , mIgnoreAudioOutputFormat(false)
   , mStartTime(-1)
   , mHitAudioDecodeError(false)
@@ -173,6 +176,54 @@ MediaDecoderReader::UpdateBuffered()
   mBuffered = GetBuffered();
 }
 
+void
+MediaDecoderReader::ThrottledNotifyDataArrived(const Interval<int64_t>& aInterval)
+{
+  MOZ_ASSERT(OnTaskQueue());
+  NS_ENSURE_TRUE_VOID(!mShutdown);
+
+  if (mThrottledInterval.isNothing()) {
+    mThrottledInterval.emplace(aInterval);
+  } else if (!mThrottledInterval.ref().Contiguous(aInterval)) {
+    DoThrottledNotify();
+    mThrottledInterval.emplace(aInterval);
+  } else {
+    mThrottledInterval = Some(mThrottledInterval.ref().Span(aInterval));
+  }
+
+  // If it's been long enough since our last update, do it.
+  if (TimeStamp::Now() - mLastThrottledNotify > mThrottleDuration) {
+    DoThrottledNotify();
+  } else if (!mThrottledNotify.Exists()) {
+    // Otherwise, schedule an update if one isn't scheduled already.
+    nsRefPtr<MediaDecoderReader> self = this;
+    mThrottledNotify.Begin(
+      mTimer->WaitUntil(mLastThrottledNotify + mThrottleDuration, __func__)
+      ->Then(TaskQueue(), __func__,
+             [self] () -> void {
+               self->mThrottledNotify.Complete();
+               NS_ENSURE_TRUE_VOID(!self->mShutdown);
+               self->DoThrottledNotify();
+             },
+             [self] () -> void {
+               self->mThrottledNotify.Complete();
+               NS_WARNING("throttle callback rejected");
+             })
+    );
+  }
+}
+
+void
+MediaDecoderReader::DoThrottledNotify()
+{
+  MOZ_ASSERT(OnTaskQueue());
+  mLastThrottledNotify = TimeStamp::Now();
+  mThrottledNotify.DisconnectIfExists();
+  Interval<int64_t> interval = mThrottledInterval.ref();
+  mThrottledInterval.reset();
+  NotifyDataArrived(interval);
+}
+
 media::TimeIntervals
 MediaDecoderReader::GetBuffered()
 {
@@ -362,6 +413,8 @@ MediaDecoderReader::Shutdown()
   mBaseAudioPromise.RejectIfExists(END_OF_STREAM, __func__);
   mBaseVideoPromise.RejectIfExists(END_OF_STREAM, __func__);
 
+  mThrottledNotify.DisconnectIfExists();
+
   ReleaseMediaResources();
   mDuration.DisconnectIfConnected();
   mBuffered.DisconnectAll();
@@ -383,6 +436,7 @@ MediaDecoderReader::Shutdown()
     p = ShutdownPromise::CreateAndResolve(true, __func__);
   }
 
+  mTimer = nullptr;
   mDecoder = nullptr;
 
   return p;

dom/media/MediaDecoderReader.h

@@ -11,7 +11,9 @@
 #include "MediaData.h"
 #include "MediaPromise.h"
 #include "MediaQueue.h"
+#include "MediaTimer.h"
 #include "AudioCompactor.h"
+#include "Intervals.h"
 #include "TimeUnits.h"
 
 namespace mozilla {
@@ -247,19 +249,32 @@ public:
 protected:
   friend class TrackBuffer;
   virtual void NotifyDataArrivedInternal(uint32_t aLength, int64_t aOffset) { }
-  void NotifyDataArrived(uint32_t aLength, int64_t aOffset)
+
+  void NotifyDataArrived(const media::Interval<int64_t>& aInfo)
   {
     MOZ_ASSERT(OnTaskQueue());
     NS_ENSURE_TRUE_VOID(!mShutdown);
-    NotifyDataArrivedInternal(aLength, aOffset);
+    NotifyDataArrivedInternal(aInfo.Length(), aInfo.mStart);
     UpdateBuffered();
   }
 
+  // Invokes NotifyDataArrived while throttling the calls to occur at most every mThrottleDuration ms.
+  void ThrottledNotifyDataArrived(const media::Interval<int64_t>& aInterval);
+  void DoThrottledNotify();
+
 public:
-  void DispatchNotifyDataArrived(uint32_t aLength, int64_t aOffset)
+  // In situations where these notifications come from stochastic network
+  // activity, we can save significant recomputation by throttling the delivery
+  // of these updates to the reader implementation. We don't want to do this
+  // throttling when the update comes from MSE code, since that code needs the
+  // updates to be observable immediately, and is generally less
+  // trigger-happy with notifications anyway.
+  void DispatchNotifyDataArrived(uint32_t aLength, int64_t aOffset, bool aThrottleUpdates)
   {
     RefPtr<nsRunnable> r =
-      NS_NewRunnableMethodWithArgs<uint32_t, int64_t>(this, &MediaDecoderReader::NotifyDataArrived, aLength, aOffset);
+      NS_NewRunnableMethodWithArg<media::Interval<int64_t>>(this, aThrottleUpdates ? &MediaDecoderReader::ThrottledNotifyDataArrived
+                                                                                   : &MediaDecoderReader::NotifyDataArrived,
+                                                            media::Interval<int64_t>(aOffset, aOffset + aLength));
     TaskQueue()->Dispatch(r.forget(), AbstractThread::DontAssertDispatchSuccess);
   }
 
@@ -358,6 +373,9 @@ protected:
   // State-watching manager.
   WatchManager<MediaDecoderReader> mWatchManager;
 
+  // MediaTimer.
+  nsRefPtr<MediaTimer> mTimer;
+
   // Buffered range.
   Canonical<media::TimeIntervals> mBuffered;
 public:
@@ -370,6 +388,12 @@ protected:
   // Duration, mirrored from the state machine task queue.
   Mirror<media::NullableTimeUnit> mDuration;
 
+  // State for ThrottledNotifyDataArrived.
+  MediaPromiseRequestHolder<MediaTimerPromise> mThrottledNotify;
+  const TimeDuration mThrottleDuration;
+  TimeStamp mLastThrottledNotify;
+  Maybe<media::Interval<int64_t>> mThrottledInterval;
+
   // Whether we should accept media that we know we can't play
   // directly, because they have a number of channel higher than
   // what we support.

dom/media/MediaDecoderStateMachine.h

@@ -266,9 +266,9 @@ public:
     return 0;
   }
 
-  void DispatchNotifyDataArrived(uint32_t aLength, int64_t aOffset)
+  void DispatchNotifyDataArrived(uint32_t aLength, int64_t aOffset, bool aThrottleUpdates)
   {
-    mReader->DispatchNotifyDataArrived(aLength, aOffset);
+    mReader->DispatchNotifyDataArrived(aLength, aOffset, aThrottleUpdates);
   }
 
   AbstractCanonical<media::TimeIntervals>* CanonicalBuffered() { return mReader->CanonicalBuffered(); }

dom/media/MediaResource.cpp

@@ -463,7 +463,8 @@ ChannelMediaResource::CopySegmentToCache(nsIInputStream *aInStream,
 {
   CopySegmentClosure* closure = static_cast<CopySegmentClosure*>(aClosure);
 
-  closure->mResource->mDecoder->NotifyDataArrived(aCount, closure->mResource->mOffset);
+  closure->mResource->mDecoder->NotifyDataArrived(aCount, closure->mResource->mOffset,
+                                                  /* aThrottleUpdates = */ true);
 
   // Keep track of where we're up to.
   RESOURCE_LOG("%p [ChannelMediaResource]: CopySegmentToCache at mOffset [%lld] add "

dom/media/mediasource/SourceBuffer.cpp

@@ -303,7 +303,7 @@ SourceBuffer::Ended()
   mContentManager->Ended();
   // We want the MediaSourceReader to refresh its buffered range as it may
   // have been modified (end lined up).
-  mMediaSource->GetDecoder()->NotifyDataArrived(1, mReportedOffset++);
+  mMediaSource->GetDecoder()->NotifyDataArrived(1, mReportedOffset++, /* aThrottleUpdates = */ false);
 }
 
 SourceBuffer::SourceBuffer(MediaSource* aMediaSource, const nsACString& aType)
@@ -498,7 +498,7 @@ SourceBuffer::AppendDataCompletedWithSuccess(bool aHasActiveTracks)
     // Tell our parent decoder that we have received new data.
     // The information provided do not matter much so long as it is monotonically
     // increasing.
-    mMediaSource->GetDecoder()->NotifyDataArrived(1, mReportedOffset++);
+    mMediaSource->GetDecoder()->NotifyDataArrived(1, mReportedOffset++, /* aThrottleUpdates = */ false);
     // Send progress event.
     mMediaSource->GetDecoder()->NotifyBytesDownloaded();
   }

dom/media/mediasource/SourceBufferDecoder.cpp

@@ -197,7 +197,7 @@ SourceBufferDecoder::GetOwner()
 }
 
 void
-SourceBufferDecoder::NotifyDataArrived(uint32_t aLength, int64_t aOffset)
+SourceBufferDecoder::NotifyDataArrived(uint32_t aLength, int64_t aOffset, bool aThrottleUpdates)
 {
 }
 

dom/media/mediasource/SourceBufferDecoder.h

@@ -47,7 +47,7 @@ public:
   virtual void FirstFrameLoaded(nsAutoPtr<MediaInfo> aInfo,
                                 MediaDecoderEventVisibility aEventVisibility) final override;
   virtual void NotifyBytesConsumed(int64_t aBytes, int64_t aOffset) final override;
-  virtual void NotifyDataArrived(uint32_t aLength, int64_t aOffset) final override;
+  virtual void NotifyDataArrived(uint32_t aLength, int64_t aOffset, bool aThrottleUpdates) final override;
   virtual void NotifyDecodedFrames(uint32_t aParsed, uint32_t aDecoded, uint32_t aDropped) final override;
   virtual void NotifyWaitingForResourcesStatusChanged() final override;
   virtual void OnReadMetadataCompleted() final override;

dom/media/mediasource/TrackBuffer.cpp

@@ -294,8 +294,7 @@ nsRefPtr<TrackBuffer::BufferedRangesUpdatedPromise>
 TrackBuffer::UpdateBufferedRanges(Interval<int64_t> aByteRange, bool aNotifyParent)
 {
   if (aByteRange.Length()) {
-    mCurrentDecoder->GetReader()->NotifyDataArrived(uint32_t(aByteRange.Length()),
-                                                    aByteRange.mStart);
+    mCurrentDecoder->GetReader()->NotifyDataArrived(aByteRange);
   }
 
   // Recalculate and cache our new buffered range.
@@ -325,7 +324,7 @@ TrackBuffer::UpdateBufferedRanges(Interval<int64_t> aByteRange, bool aNotifyPare
         // specific SourceBufferDecoder's data stream.  Pass bogus values here to
         // force parent decoder's state machine to recompute end time for
         // infinite length media.
-        parent->NotifyDataArrived(0, 0);
+        parent->NotifyDataArrived(0, 0, /* aThrottleUpdates = */ false);
         parent->NotifyBytesDownloaded();
       });
     AbstractThread::MainThread()->Dispatch(task.forget());
@@ -762,8 +761,7 @@ TrackBuffer::InitializeDecoder(SourceBufferDecoder* aDecoder)
   MSE_DEBUG("Initializing subdecoder %p reader %p",
             aDecoder, reader);
 
-  reader->NotifyDataArrived(uint32_t(mLastAppendRange.Length()),
-                            mLastAppendRange.mStart);
+  reader->NotifyDataArrived(mLastAppendRange);
 
   // HACK WARNING:
   // We only reach this point once we know that we have a complete init segment.

dom/media/omx/MediaOmxReader.cpp

@@ -96,7 +96,7 @@ private:
 
     while (mLength) {
       uint32_t length = std::min<uint64_t>(mLength, UINT32_MAX);
-      mOmxReader->NotifyDataArrived(length, mOffset);
+      mOmxReader->NotifyDataArrived(Interval<int64_t>(mOffset, mOffset + length));
       mLength -= length;
       mOffset += length;
     }

dom/media/webaudio/BufferDecoder.cpp

@@ -153,12 +153,6 @@ BufferDecoder::NotifyWaitingForResourcesStatusChanged()
   // ignore
 }
 
-void
-BufferDecoder::NotifyDataArrived(uint32_t aLength, int64_t aOffset)
-{
-  // ignore
-}
-
 MediaDecoderOwner*
 BufferDecoder::GetOwner()
 {

dom/media/webaudio/BufferDecoder.h

@@ -69,7 +69,7 @@ public:
 
   virtual void NotifyWaitingForResourcesStatusChanged() final override;
 
-  virtual void NotifyDataArrived(uint32_t aLength, int64_t aOffset) final override;
+  virtual void NotifyDataArrived(uint32_t, int64_t, bool) final override {};
 
 private:
   virtual ~BufferDecoder();