Bug 1795507 - part3 : make output data return promise and simplify the threading for the video stream. r=jolin

This patch further simplify the logic of returning decoded data by
making it async.

As now we don't need to return decoded data on different thread than the
task queue, it also makes the threading for the video stream easier.

Differential Revision: https://phabricator.services.mozilla.com/D161550

dom/media/platforms/wmf/MFMediaEngineAudioStream.cpp

@@ -75,12 +75,11 @@ bool MFMediaEngineAudioStream::HasEnoughRawData() const {
   return mRawDataQueueForFeedingEngine.Duration() >= AMPLE_AUDIO_USECS;
 }
 
-already_AddRefed<MediaData> MFMediaEngineAudioStream::OutputData() {
+already_AddRefed<MediaData> MFMediaEngineAudioStream::OutputDataInternal() {
+  AssertOnTaskQueue();
   if (mRawDataQueueForGeneratingOutput.GetSize() == 0) {
-    LOGV("Hasn't got raw data for generating output yet");
     return nullptr;
   }
-
   // The media engine doesn't provide a way to allow us to access decoded audio
   // frames, and the audio playback will be handled internally inside the media
   // engine. So we simply return fake audio data.

dom/media/platforms/wmf/MFMediaEngineAudioStream.h

@@ -27,14 +27,14 @@ class MFMediaEngineAudioStream final : public MFMediaEngineStream {
     return TrackInfo::TrackType::kAudioTrack;
   }
 
-  already_AddRefed<MediaData> OutputData() override;
-
  private:
   HRESULT CreateMediaType(const TrackInfo& aInfo,
                           IMFMediaType** aMediaType) override;
 
   bool HasEnoughRawData() const override;
 
+  already_AddRefed<MediaData> OutputDataInternal() override;
+
   // For MF_MT_USER_DATA. Currently only used for AAC.
   nsTArray<BYTE> mAACUserData;
 

dom/media/platforms/wmf/MFMediaEngineStream.cpp

@@ -51,11 +51,8 @@ RefPtr<MediaDataDecoder::DecodePromise> MFMediaEngineStreamWrapper::Decode(
   Unused << mTaskQueue->Dispatch(NS_NewRunnableFunction(
       "MFMediaEngineStreamWrapper::Decode",
       [sample = RefPtr{aSample}, stream]() { stream->NotifyNewData(sample); }));
-
-  RefPtr<MediaData> outputData = mStream->OutputData();
-  return outputData ? DecodePromise::CreateAndResolve(
-                          DecodedData{std::move(outputData)}, __func__)
-                    : DecodePromise::CreateAndResolve(DecodedData{}, __func__);
+  return InvokeAsync(mTaskQueue, mStream.Get(), __func__,
+                     &MFMediaEngineStream::OutputData);
 }
 
 RefPtr<MediaDataDecoder::DecodePromise> MFMediaEngineStreamWrapper::Drain() {
@@ -65,14 +62,8 @@ RefPtr<MediaDataDecoder::DecodePromise> MFMediaEngineStreamWrapper::Drain() {
         MediaResult(NS_ERROR_FAILURE, "MFMediaEngineStreamWrapper is shutdown"),
         __func__);
   }
-
-  DecodedData outputs;
-  RefPtr<MediaData> outputData = mStream->OutputData();
-  while (outputData) {
-    outputs.AppendElement(outputData);
-    outputData = mStream->OutputData();
-  }
-  return DecodePromise::CreateAndResolve(std::move(outputs), __func__);
+  return InvokeAsync(mTaskQueue, mStream.Get(), __func__,
+                     &MFMediaEngineStream::Drain);
 }
 
 RefPtr<MediaDataDecoder::FlushPromise> MFMediaEngineStreamWrapper::Flush() {
@@ -220,6 +211,7 @@ void MFMediaEngineStream::Shutdown() {
         self->mParentSource = nullptr;
         self->mRawDataQueueForFeedingEngine.Reset();
         self->mRawDataQueueForGeneratingOutput.Reset();
+        self->ShutdownCleanUpOnTaskQueue();
       }));
 }
 
@@ -265,10 +257,7 @@ IFACEMETHODIMP MFMediaEngineStream::RequestSample(IUnknown* aToken) {
         SLOGV("RequestSample, token amount=%zu", mSampleRequestTokens.size());
         ReplySampleRequestIfPossible();
         if (!HasEnoughRawData() && mParentSource && !IsEnded()) {
-          SLOGV("Dispatch a sample request, queue duration=%" PRId64,
-                mRawDataQueueForFeedingEngine.Duration());
-          mParentSource->mRequestSampleEvent.Notify(
-              SampleRequest{TrackType(), false /* isEnough */});
+          SendRequestSampleEvent(false /* isEnough */);
         }
       }));
   return S_OK;
@@ -410,7 +399,7 @@ void MFMediaEngineStream::NotifyNewData(MediaRawData* aSample) {
   if (IsShutdown()) {
     return;
   }
-  bool wasEnough = HasEnoughRawData();
+  const bool wasEnough = HasEnoughRawData();
   mRawDataQueueForFeedingEngine.Push(aSample);
   mRawDataQueueForGeneratingOutput.Push(aSample);
   SLOGV("NotifyNewData, push data [%" PRId64 ", %" PRId64
@@ -424,12 +413,19 @@ void MFMediaEngineStream::NotifyNewData(MediaRawData* aSample) {
   }
   ReplySampleRequestIfPossible();
   if (!wasEnough && HasEnoughRawData()) {
-    SLOGV("data is enough");
-    mParentSource->mRequestSampleEvent.Notify(
-        SampleRequest{TrackType(), true /* isEnough */});
+    SendRequestSampleEvent(true /* isEnough */);
   }
 }
 
+void MFMediaEngineStream::SendRequestSampleEvent(bool aIsEnough) {
+  AssertOnTaskQueue();
+  SLOGV("data is %s, queue duration=%" PRId64,
+        aIsEnough ? "enough" : "not enough",
+        mRawDataQueueForFeedingEngine.Duration());
+  mParentSource->mRequestSampleEvent.Notify(
+      SampleRequest{TrackType(), aIsEnough});
+}
+
 void MFMediaEngineStream::NotifyEndOfStreamInternal() {
   AssertOnTaskQueue();
   if (mReceivedEOS) {
@@ -460,6 +456,32 @@ RefPtr<MediaDataDecoder::FlushPromise> MFMediaEngineStream::Flush() {
   return MediaDataDecoder::FlushPromise::CreateAndResolve(true, __func__);
 }
 
+RefPtr<MediaDataDecoder::DecodePromise> MFMediaEngineStream::OutputData() {
+  AssertOnTaskQueue();
+  MediaDataDecoder::DecodedData outputs;
+  if (RefPtr<MediaData> outputData = OutputDataInternal()) {
+    outputs.AppendElement(outputData);
+    SLOGV("Output data [%" PRId64 ",%" PRId64 "]",
+          outputData->mTime.ToMicroseconds(),
+          outputData->GetEndTime().ToMicroseconds());
+  }
+  return MediaDataDecoder::DecodePromise::CreateAndResolve(std::move(outputs),
+                                                           __func__);
+};
+
+RefPtr<MediaDataDecoder::DecodePromise> MFMediaEngineStream::Drain() {
+  AssertOnTaskQueue();
+  MediaDataDecoder::DecodedData outputs;
+  while (RefPtr<MediaData> outputData = OutputDataInternal()) {
+    outputs.AppendElement(outputData);
+    SLOGV("Output data [%" PRId64 ",%" PRId64 "]",
+          outputData->mTime.ToMicroseconds(),
+          outputData->GetEndTime().ToMicroseconds());
+  }
+  return MediaDataDecoder::DecodePromise::CreateAndResolve(std::move(outputs),
+                                                           __func__);
+}
+
 void MFMediaEngineStream::AssertOnTaskQueue() const {
   MOZ_ASSERT(mTaskQueue->IsCurrentThreadIn());
 }

dom/media/platforms/wmf/MFMediaEngineStream.h

@@ -92,9 +92,9 @@ class MFMediaEngineStream
 
   virtual MFMediaEngineVideoStream* AsVideoStream() { return nullptr; }
 
-  // Overwrite this method to support returning decoded data. Called on
-  // MediaPDecoder thread (wrapper thread).
-  virtual already_AddRefed<MediaData> OutputData() = 0;
+  RefPtr<MediaDataDecoder::DecodePromise> OutputData();
+
+  virtual RefPtr<MediaDataDecoder::DecodePromise> Drain();
 
   virtual MediaDataDecoder::ConversionRequired NeedsConversion() const {
     return MediaDataDecoder::ConversionRequired::kNeedNone;
@@ -120,6 +120,16 @@ class MFMediaEngineStream
 
   bool IsEnded() const;
 
+  // Overwrite this method if inherited class needs to perform clean up on the
+  // task queue when the stream gets shutdowned.
+  virtual void ShutdownCleanUpOnTaskQueue(){};
+
+  // Inherited class must implement this method to return decoded data. it
+  // should uses `mRawDataQueueForGeneratingOutput` to generate output.
+  virtual already_AddRefed<MediaData> OutputDataInternal() = 0;
+
+  void SendRequestSampleEvent(bool aIsEnough);
+
   void AssertOnTaskQueue() const;
   void AssertOnMFThreadPool() const;
 

dom/media/platforms/wmf/MFMediaEngineVideoStream.cpp

@@ -34,15 +34,33 @@ MFMediaEngineVideoStream* MFMediaEngineVideoStream::Create(
   return stream;
 }
 
+void MFMediaEngineVideoStream::SetKnowsCompositor(
+    layers::KnowsCompositor* aKnowsCompositor) {
+  ComPtr<MFMediaEngineVideoStream> self = this;
+  Unused << mTaskQueue->Dispatch(NS_NewRunnableFunction(
+      "MFMediaEngineStream::SetKnowsCompositor",
+      [self, knowCompositor = RefPtr<layers::KnowsCompositor>{aKnowsCompositor},
+       this]() {
+        mKnowsCompositor = knowCompositor;
+        LOGV("Set SetKnowsCompositor=%p", mKnowsCompositor.get());
+        ResolvePendingDrainPromiseIfNeeded();
+      }));
+}
+
 void MFMediaEngineVideoStream::SetDCompSurfaceHandle(
     HANDLE aDCompSurfaceHandle) {
-  MutexAutoLock lock(mMutex);
-  if (mDCompSurfaceHandle == aDCompSurfaceHandle) {
-    return;
-  }
-  mDCompSurfaceHandle = aDCompSurfaceHandle;
-  mNeedRecreateImage = true;
-  LOGV("Set DCompSurfaceHandle, handle=%p", mDCompSurfaceHandle);
+  ComPtr<MFMediaEngineVideoStream> self = this;
+  Unused << mTaskQueue->Dispatch(NS_NewRunnableFunction(
+      "MFMediaEngineStream::SetDCompSurfaceHandle",
+      [self, aDCompSurfaceHandle, this]() {
+        if (mDCompSurfaceHandle == aDCompSurfaceHandle) {
+          return;
+        }
+        mDCompSurfaceHandle = aDCompSurfaceHandle;
+        mNeedRecreateImage = true;
+        LOGV("Set DCompSurfaceHandle, handle=%p", mDCompSurfaceHandle);
+        ResolvePendingDrainPromiseIfNeeded();
+      }));
 }
 
 HRESULT MFMediaEngineVideoStream::CreateMediaType(const TrackInfo& aInfo,
@@ -172,25 +190,20 @@ bool MFMediaEngineVideoStream::HasEnoughRawData() const {
   return mRawDataQueueForFeedingEngine.Duration() >= VIDEO_VIDEO_USECS;
 }
 
-already_AddRefed<MediaData> MFMediaEngineVideoStream::OutputData() {
-  MutexAutoLock lock(mMutex);
-
+bool MFMediaEngineVideoStream::IsDCompImageReady() {
+  AssertOnTaskQueue();
   if (!mDCompSurfaceHandle || mDCompSurfaceHandle == INVALID_HANDLE_VALUE) {
     LOGV("Can't create image without a valid dcomp surface handle");
-    return nullptr;
+    return false;
   }
 
   if (!mKnowsCompositor) {
     LOGV("Can't create image without the knows compositor");
-    return nullptr;
-  }
-
-  if (mRawDataQueueForGeneratingOutput.GetSize() == 0) {
-    LOGV("Hasn't got raw data for generating output yet");
-    return nullptr;
+    return false;
   }
 
   if (!mDcompSurfaceImage || mNeedRecreateImage) {
+    MutexAutoLock lock(mMutex);
     // DirectComposition only supports RGBA. We use DXGI_FORMAT_B8G8R8A8_UNORM
     // as a default because we can't know what format the dcomp surface is.
     // https://docs.microsoft.com/en-us/windows/win32/api/dcomp/nf-dcomp-idcompositionsurfacefactory-createsurface
@@ -201,11 +214,51 @@ already_AddRefed<MediaData> MFMediaEngineVideoStream::OutputData() {
     LOGV("Created dcomp surface image, handle=%p, size=[%u,%u]",
          mDCompSurfaceHandle, mDisplay.Width(), mDisplay.Height());
   }
+  return true;
+}
 
+already_AddRefed<MediaData> MFMediaEngineVideoStream::OutputDataInternal() {
+  AssertOnTaskQueue();
+  if (mRawDataQueueForGeneratingOutput.GetSize() == 0 || !IsDCompImageReady()) {
+    return nullptr;
+  }
   RefPtr<MediaRawData> sample = mRawDataQueueForGeneratingOutput.PopFront();
-  return VideoData::CreateFromImage(mDisplay, sample->mOffset, sample->mTime,
-                                    sample->mDuration, mDcompSurfaceImage,
-                                    sample->mKeyframe, sample->mTimecode);
+  RefPtr<VideoData> output;
+  {
+    MutexAutoLock lock(mMutex);
+    output = VideoData::CreateFromImage(
+        mDisplay, sample->mOffset, sample->mTime, sample->mDuration,
+        mDcompSurfaceImage, sample->mKeyframe, sample->mTimecode);
+  }
+  return output.forget();
+}
+
+RefPtr<MediaDataDecoder::DecodePromise> MFMediaEngineVideoStream::Drain() {
+  AssertOnTaskQueue();
+  MediaDataDecoder::DecodedData outputs;
+  if (!IsDCompImageReady()) {
+    return mPendingDrainPromise.Ensure(__func__);
+  }
+  return MFMediaEngineStream::Drain();
+}
+
+void MFMediaEngineVideoStream::ResolvePendingDrainPromiseIfNeeded() {
+  AssertOnTaskQueue();
+  if (mPendingDrainPromise.IsEmpty()) {
+    return;
+  }
+  if (!IsDCompImageReady()) {
+    return;
+  }
+  MediaDataDecoder::DecodedData outputs;
+  while (RefPtr<MediaData> outputData = OutputDataInternal()) {
+    outputs.AppendElement(outputData);
+    LOGV("Output data [%" PRId64 ",%" PRId64 "]",
+         outputData->mTime.ToMicroseconds(),
+         outputData->GetEndTime().ToMicroseconds());
+  }
+  mPendingDrainPromise.Resolve(std::move(outputs), __func__);
+  LOGV("Resolved pending drain promise");
 }
 
 MediaDataDecoder::ConversionRequired MFMediaEngineVideoStream::NeedsConversion()
@@ -251,6 +304,11 @@ void MFMediaEngineVideoStream::UpdateConfig(const VideoInfo& aInfo) {
       MEStreamFormatChanged, GUID_NULL, S_OK, mediaType.Get()));
 }
 
+void MFMediaEngineVideoStream::ShutdownCleanUpOnTaskQueue() {
+  AssertOnTaskQueue();
+  mPendingDrainPromise.RejectIfExists(NS_ERROR_DOM_MEDIA_CANCELED, __func__);
+}
+
 #undef LOGV
 
 }  // namespace mozilla

dom/media/platforms/wmf/MFMediaEngineVideoStream.h

@@ -35,21 +35,12 @@ class MFMediaEngineVideoStream final : public MFMediaEngineStream {
     return TrackInfo::TrackType::kVideoTrack;
   }
 
-  void SetKnowsCompositor(layers::KnowsCompositor* aKnowsCompositor) {
-    MutexAutoLock lock(mMutex);
-    mKnowsCompositor = aKnowsCompositor;
-  }
+  void SetKnowsCompositor(layers::KnowsCompositor* aKnowsCompositor);
 
-  HANDLE GetDcompSurfaceHandle() {
-    MutexAutoLock lock(mMutex);
-    return mDCompSurfaceHandle;
-  }
   void SetDCompSurfaceHandle(HANDLE aDCompSurfaceHandle);
 
   MFMediaEngineVideoStream* AsVideoStream() override { return this; }
 
-  already_AddRefed<MediaData> OutputData() override;
-
   MediaDataDecoder::ConversionRequired NeedsConversion() const override;
 
   // Called by MFMediaEngineParent when we are creating a video decoder for
@@ -57,6 +48,8 @@ class MFMediaEngineVideoStream final : public MFMediaEngineStream {
   // change happens during playback.
   void SetConfig(const TrackInfo& aConfig);
 
+  RefPtr<MediaDataDecoder::DecodePromise> Drain() override;
+
  private:
   HRESULT
   CreateMediaType(const TrackInfo& aInfo, IMFMediaType** aMediaType) override;
@@ -65,11 +58,20 @@ class MFMediaEngineVideoStream final : public MFMediaEngineStream {
 
   void UpdateConfig(const VideoInfo& aInfo);
 
-  Mutex mMutex{"MFMediaEngineVideoStream"};
+  already_AddRefed<MediaData> OutputDataInternal() override;
 
-  HANDLE mDCompSurfaceHandle MOZ_GUARDED_BY(mMutex);
-  bool mNeedRecreateImage MOZ_GUARDED_BY(mMutex);
-  RefPtr<layers::KnowsCompositor> mKnowsCompositor MOZ_GUARDED_BY(mMutex);
+  bool IsDCompImageReady();
+
+  void ResolvePendingDrainPromiseIfNeeded();
+
+  void ShutdownCleanUpOnTaskQueue() override;
+
+  // Task queue only members.
+  HANDLE mDCompSurfaceHandle;
+  bool mNeedRecreateImage;
+  RefPtr<layers::KnowsCompositor> mKnowsCompositor;
+
+  Mutex mMutex{"MFMediaEngineVideoStream"};
   gfx::IntSize mDisplay MOZ_GUARDED_BY(mMutex);
 
   // Set on the initialization, won't be changed after that.
@@ -83,6 +85,12 @@ class MFMediaEngineVideoStream final : public MFMediaEngineStream {
   // this flag to true, then we know any config being set afterward indicating
   // a new config change.
   bool mHasReceivedInitialCreateDecoderConfig;
+
+  // When draining, the track should return all decoded data. However, if the
+  // dcomp image hasn't been ready yet, then we won't have any decoded data to
+  // return. This promise is used for that case, and will be resolved once we
+  // have dcomp image.
+  MozPromiseHolder<MediaDataDecoder::DecodePromise> mPendingDrainPromise;
 };
 
 }  // namespace mozilla