Bug 1404997 - P20. Make MediaStreamListener::NotifyPull asynchronous. r=padenot

The operations is done in two ways:
1- Process all the MediaStreamListener at once, which returns a promise that will be resolved once the operation is completed.
2- As the Cubeb audio callback must be resolved immediately, the MSG will wait for all the promises to be resolved until it continues the operation of feeding the callback the necessary data.

This will allow to parallelize the stream's tracks' audio decoding.

MozReview-Commit-ID: EeoDvxnJyWV

--HG--
extra : rebase_source : 3d09af5aa3c80c4892a4d9af80842541d8fc33bb

dom/media/MediaStreamGraph.cpp

@@ -1149,12 +1149,25 @@ MediaStreamGraphImpl::UpdateGraph(GraphTime aEndBlockingDecisions)
   bool ensureNextIteration = false;
 
   // Grab pending stream input and compute blocking time
+  // TODO: Ensure that heap memory allocations isn't going to be a problem.
+  // Maybe modify code to use nsAutoTArray as out parameters.
+  nsTArray<RefPtr<SourceMediaStream::NotifyPullPromise>> promises;
+  for (MediaStream* stream : mStreams) {
+    if (SourceMediaStream* is = stream->AsSourceStream()) {
+      promises.AppendElements(
+        is->PullNewData(aEndBlockingDecisions, &ensureNextIteration));
+    }
+  }
+
+  // Wait until all PullEnabled stream's listeners have completed.
+  if (!promises.IsEmpty()) {
+    AwaitAll(promises);
+  }
+
   for (MediaStream* stream : mStreams) {
     if (SourceMediaStream* is = stream->AsSourceStream()) {
-      is->PullNewData(aEndBlockingDecisions, &ensureNextIteration);
       is->ExtractPendingInput();
     }
-
     if (stream->mFinished) {
       // The stream's not suspended, and since it's finished, underruns won't
       // stop it playing out. So there's no blocking other than what we impose
@@ -2694,43 +2707,48 @@ SourceMediaStream::SetPullEnabled(bool aEnabled)
   }
 }
 
-void
+nsTArray<RefPtr<SourceMediaStream::NotifyPullPromise>>
 SourceMediaStream::PullNewData(StreamTime aDesiredUpToTime,
                                bool* aEnsureNextIteration)
 {
+  // 2 is the average number of listeners per SourceMediaStream.
+  nsTArray<RefPtr<SourceMediaStream::NotifyPullPromise>> promises(2);
   MutexAutoLock lock(mMutex);
-  if (mPullEnabled && !mFinished && !mListeners.IsEmpty()) {
-    // Compute how much stream time we'll need assuming we don't block
-    // the stream at all.
-    StreamTime t = GraphTimeToStreamTime(aDesiredUpToTime);
-    StreamTime current = mTracks.GetEnd();
-    LOG(LogLevel::Verbose,
-        ("Calling NotifyPull aStream=%p t=%f current end=%f",
-          this,
-          GraphImpl()->MediaTimeToSeconds(t),
-          GraphImpl()->MediaTimeToSeconds(current)));
-    if (t > current) {
-      *aEnsureNextIteration = true;
+  if (!mPullEnabled || mFinished) {
+    return promises;
+  }
+  // Compute how much stream time we'll need assuming we don't block
+  // the stream at all.
+  StreamTime t = GraphTimeToStreamTime(aDesiredUpToTime);
+  StreamTime current = mTracks.GetEnd();
+  LOG(LogLevel::Verbose,
+      ("Calling NotifyPull aStream=%p t=%f current end=%f",
+        this,
+        GraphImpl()->MediaTimeToSeconds(t),
+        GraphImpl()->MediaTimeToSeconds(current)));
+  if (t <= current) {
+    return promises;
+  }
+  *aEnsureNextIteration = true;
 #ifdef DEBUG
-      if (mListeners.Length() == 0) {
-        LOG(
-          LogLevel::Error,
-          ("No listeners in NotifyPull aStream=%p desired=%f current end=%f",
-            this,
-            GraphImpl()->MediaTimeToSeconds(t),
-            GraphImpl()->MediaTimeToSeconds(current)));
-        DumpTrackInfo();
-      }
+  if (mListeners.Length() == 0) {
+    LOG(
+      LogLevel::Error,
+      ("No listeners in NotifyPull aStream=%p desired=%f current end=%f",
+        this,
+        GraphImpl()->MediaTimeToSeconds(t),
+        GraphImpl()->MediaTimeToSeconds(current)));
+    DumpTrackInfo();
+  }
 #endif
-      for (uint32_t j = 0; j < mListeners.Length(); ++j) {
-        MediaStreamListener* l = mListeners[j];
-        {
-          MutexAutoUnlock unlock(mMutex);
-          l->NotifyPull(GraphImpl(), t);
-        }
-      }
+  for (uint32_t j = 0; j < mListeners.Length(); ++j) {
+    MediaStreamListener* l = mListeners[j];
+    {
+      MutexAutoUnlock unlock(mMutex);
+      promises.AppendElement(l->AsyncNotifyPull(GraphImpl(), t));
     }
   }
+  return promises;
 }
 
 void

dom/media/MediaStreamGraph.h

@@ -12,6 +12,7 @@
 #include "StreamTracks.h"
 #include "VideoSegment.h"
 #include "mozilla/LinkedList.h"
+#include "mozilla/MozPromise.h"
 #include "mozilla/Mutex.h"
 #include "mozilla/TaskQueue.h"
 #include "nsAutoPtr.h"
@@ -702,7 +703,9 @@ public:
    * Call all MediaStreamListeners to request new data via the NotifyPull API
    * (if enabled).
    */
-  void PullNewData(StreamTime aDesiredUpToTime, bool* aEnsureNextIteration);
+  typedef MozPromise<bool, bool, true /* is exclusive */ > NotifyPullPromise;
+  nsTArray<RefPtr<NotifyPullPromise>> PullNewData(StreamTime aDesiredUpToTime,
+                                                  bool* aEnsureNextIteration);
 
   /**
    * Extract any state updates pending in the stream, and apply them.

dom/media/MediaStreamListener.h

@@ -39,7 +39,8 @@ class VideoSegment;
  * callback to notify of the initial blocking state. Also, if a listener is
  * attached to a stream that has already finished, we'll call NotifyFinished.
  */
-class MediaStreamListener {
+class MediaStreamListener
+{
 protected:
   // Protected destructor, to discourage deletion outside of Release():
   virtual ~MediaStreamListener() {}
@@ -60,6 +61,14 @@ public:
    * some reason, then data before aDesiredTime may not be played immediately.
    */
   virtual void NotifyPull(MediaStreamGraph* aGraph, StreamTime aDesiredTime) {}
+  virtual RefPtr<SourceMediaStream::NotifyPullPromise> AsyncNotifyPull(
+    MediaStreamGraph* aGraph,
+    StreamTime aDesiredTime)
+  {
+    NotifyPull(aGraph, aDesiredTime);
+    return SourceMediaStream::NotifyPullPromise::CreateAndResolve(true,
+                                                                  __func__);
+  }
 
   enum Blocking {
     BLOCKED,