Audio|VideoDecoderBroker: more weakptr, keep callbacks on main thread.

Using unretained for decoder/selector callbacks is generally not safe /
fragile. For ex, some decoders (e.g. those that offload) will call the
output callback after destruction. Another example: decoder selector
internally posts the Select() callback, so it may run after destruction.

Additionally, sending incoming callbacks as CrossThreadFunctions for
invoking on the media thread is not safe. We may destruct while some
callbacks are still pending, destroying the callbacks internal state
(including bindings to blink GC types) off the main thread. Now we
ensure that all callbacks are saved on the main thread and we extend
the "client" interfaces of the broker for proxying the callbacks safely.

Bug: 1120431

Change-Id: I2ccfe46982cda9b18e47a56424b25f54f8877883
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2366661
Commit-Queue: Chrome Cunningham <chcunningham@chromium.org>
Auto-Submit: Chrome Cunningham <chcunningham@chromium.org>
Reviewed-by: Dan Sanders <sandersd@chromium.org>
Cr-Commit-Position: refs/heads/master@{#800776}

third_party/blink/renderer/modules/webcodecs/audio_decoder_broker.cc

@@ -4,6 +4,7 @@
 
 #include "third_party/blink/renderer/modules/webcodecs/audio_decoder_broker.h"
 
+#include <limits>
 #include <memory>
 #include <string>
 
@@ -103,49 +104,52 @@ class MediaAudioTaskWrapper {
   MediaAudioTaskWrapper(const MediaAudioTaskWrapper&) = delete;
   MediaAudioTaskWrapper& operator=(const MediaAudioTaskWrapper&) = delete;
 
-  void Initialize(const media::AudioDecoderConfig& config,
-                  CrossThreadOnceInitCB init_cb) {
+  void Initialize(const media::AudioDecoderConfig& config) {
     DVLOG(2) << __func__;
     DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
 
     selector_ = std::make_unique<WebCodecsAudioDecoderSelector>(
         media_task_runner_,
+        // TODO(chcunningham): Its ugly that we don't use a WeakPtr here, but
+        // its not possible because the callback returns non-void. It happens
+        // to be safe given the way the callback is called (never posted), but
+        // we should refactor the return to be an out-param so we can be
+        // consistent in using weak pointers.
         WTF::BindRepeating(&MediaAudioTaskWrapper::OnCreateDecoders,
                            WTF::Unretained(this)),
         WTF::BindRepeating(&MediaAudioTaskWrapper::OnDecodeOutput,
-                           WTF::Unretained(this)));
+                           weak_factory_.GetWeakPtr()));
 
     selector_->SelectDecoder(
         config, WTF::Bind(&MediaAudioTaskWrapper::OnDecoderSelected,
-                          WTF::Unretained(this), std::move(init_cb)));
+                          weak_factory_.GetWeakPtr()));
   }
 
-  void Decode(scoped_refptr<media::DecoderBuffer> buffer,
-              CrossThreadOnceDecodeCB decode_cb) {
+  void Decode(scoped_refptr<media::DecoderBuffer> buffer, int cb_id) {
     DVLOG(2) << __func__;
     DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
 
     if (!decoder_) {
-      std::move(decode_cb).Run(media::DecodeStatus::DECODE_ERROR);
+      OnDecodeDone(cb_id, media::DecodeStatus::DECODE_ERROR);
       return;
     }
 
     decoder_->Decode(std::move(buffer),
                      WTF::Bind(&MediaAudioTaskWrapper::OnDecodeDone,
-                               WTF::Unretained(this), std::move(decode_cb)));
+                               weak_factory_.GetWeakPtr(), cb_id));
   }
 
-  void Reset(CrossThreadOnceResetCB reset_cb) {
+  void Reset(int cb_id) {
     DVLOG(2) << __func__;
     DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
 
     if (!decoder_) {
-      std::move(reset_cb).Run();
+      OnReset(cb_id);
       return;
     }
 
     decoder_->Reset(WTF::Bind(&MediaAudioTaskWrapper::OnReset,
-                              WTF::Unretained(this), std::move(reset_cb)));
+                              weak_factory_.GetWeakPtr(), cb_id));
   }
 
  private:
@@ -177,8 +181,7 @@ class MediaAudioTaskWrapper {
     return audio_decoders;
   }
 
-  void OnDecoderSelected(CrossThreadOnceInitCB init_cb,
-                         std::unique_ptr<media::AudioDecoder> decoder) {
+  void OnDecoderSelected(std::unique_ptr<media::AudioDecoder> decoder) {
     DVLOG(2) << __func__;
     DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
 
@@ -200,7 +203,8 @@ class MediaAudioTaskWrapper {
     // Fire |init_cb|.
     PostCrossThreadTask(
         *main_task_runner_, FROM_HERE,
-        WTF::CrossThreadBindOnce(std::move(init_cb), status, decoder_details));
+        WTF::CrossThreadBindOnce(&CrossThreadAudioDecoderClient::OnInitialize,
+                                 weak_client_, status, decoder_details));
   }
 
   void OnDecodeOutput(scoped_refptr<media::AudioBuffer> buffer) {
@@ -213,18 +217,22 @@ class MediaAudioTaskWrapper {
                                  weak_client_, std::move(buffer)));
   }
 
-  void OnDecodeDone(CrossThreadOnceDecodeCB decode_cb,
-                    media::DecodeStatus status) {
+  void OnDecodeDone(int cb_id, media::DecodeStatus status) {
     DVLOG(2) << __func__;
     DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
-    PostCrossThreadTask(*main_task_runner_, FROM_HERE,
-                        WTF::CrossThreadBindOnce(std::move(decode_cb), status));
+    PostCrossThreadTask(
+        *main_task_runner_, FROM_HERE,
+        WTF::CrossThreadBindOnce(&CrossThreadAudioDecoderClient::OnDecodeDone,
+                                 weak_client_, cb_id, status));
   }
 
-  void OnReset(CrossThreadOnceResetCB reset_cb) {
+  void OnReset(int cb_id) {
     DVLOG(2) << __func__;
     DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
-    PostCrossThreadTask(*main_task_runner_, FROM_HERE, std::move(reset_cb));
+    PostCrossThreadTask(
+        *main_task_runner_, FROM_HERE,
+        WTF::CrossThreadBindOnce(&CrossThreadAudioDecoderClient::OnReset,
+                                 weak_client_, cb_id));
   }
 
   base::WeakPtr<CrossThreadAudioDecoderClient> weak_client_;
@@ -240,6 +248,11 @@ class MediaAudioTaskWrapper {
   media::NullMediaLog null_media_log_;
 
   SEQUENCE_CHECKER(sequence_checker_);
+
+  // Using unretained for decoder/selector callbacks is generally not safe /
+  // fragile. Some decoders (e.g. those that offload) will call the output
+  // callback after destruction.
+  base::WeakPtrFactory<MediaAudioTaskWrapper> weak_factory_{this};
 };
 
 constexpr char AudioDecoderBroker::kDefaultDisplayName[];
@@ -278,35 +291,45 @@ void AudioDecoderBroker::Initialize(const media::AudioDecoderConfig& config,
                                     const media::WaitingCB& waiting_cb) {
   DVLOG(2) << __func__;
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  DCHECK(!init_cb_) << "Initialize already pending";
 
   // The following are not currently supported in WebCodecs.
   DCHECK(!cdm_context);
   DCHECK(!waiting_cb);
 
+  init_cb_ = std::move(init_cb);
   output_cb_ = output_cb;
 
   // Clear details from previously initialized decoder. New values will arrive
   // via OnInitialize().
   decoder_details_.reset();
 
-  MediaAudioTaskWrapper::CrossThreadOnceInitCB main_loop_init_cb(
-      WTF::Bind(&AudioDecoderBroker::OnInitialize, weak_factory_.GetWeakPtr(),
-                std::move(init_cb)));
-
   PostCrossThreadTask(
       *media_task_runner_, FROM_HERE,
       WTF::CrossThreadBindOnce(&MediaAudioTaskWrapper::Initialize,
                                WTF::CrossThreadUnretained(media_tasks_.get()),
-                               config, std::move(main_loop_init_cb)));
+                               config));
 }
 
-void AudioDecoderBroker::OnInitialize(InitCB init_cb,
-                                      media::Status status,
+int AudioDecoderBroker::CreateCallbackId() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+
+  // 0 and -1 are reserved by wtf::HashMap ("empty" and "deleted").
+  while (++last_callback_id_ == 0 ||
+         last_callback_id_ == std::numeric_limits<uint32_t>::max() ||
+         pending_decode_cb_map_.Contains(last_callback_id_) ||
+         pending_reset_cb_map_.Contains(last_callback_id_))
+    ;
+
+  return last_callback_id_;
+}
+
+void AudioDecoderBroker::OnInitialize(media::Status status,
                                       base::Optional<DecoderDetails> details) {
   DVLOG(2) << __func__;
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   decoder_details_ = details;
-  std::move(init_cb).Run(status);
+  std::move(init_cb_).Run(status);
 }
 
 void AudioDecoderBroker::Decode(scoped_refptr<media::DecoderBuffer> buffer,
@@ -314,37 +337,38 @@ void AudioDecoderBroker::Decode(scoped_refptr<media::DecoderBuffer> buffer,
   DVLOG(2) << __func__;
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
 
-  MediaAudioTaskWrapper::CrossThreadOnceDecodeCB main_loop_cb(
-      WTF::Bind(&AudioDecoderBroker::OnDecodeDone, weak_factory_.GetWeakPtr(),
-                std::move(decode_cb)));
+  const int callback_id = CreateCallbackId();
+  pending_decode_cb_map_.insert(callback_id, std::move(decode_cb));
 
   PostCrossThreadTask(
       *media_task_runner_, FROM_HERE,
       WTF::CrossThreadBindOnce(&MediaAudioTaskWrapper::Decode,
                                WTF::CrossThreadUnretained(media_tasks_.get()),
-                               buffer, std::move(main_loop_cb)));
+                               buffer, callback_id));
 }
 
-void AudioDecoderBroker::OnDecodeDone(DecodeCB decode_cb,
-                                      media::DecodeStatus status) {
+void AudioDecoderBroker::OnDecodeDone(int cb_id, media::DecodeStatus status) {
   DVLOG(2) << __func__;
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
-  std::move(decode_cb).Run(status);
+  DCHECK(pending_decode_cb_map_.Contains(cb_id));
+
+  auto iter = pending_decode_cb_map_.find(cb_id);
+  std::move(iter->value).Run(status);
+  pending_decode_cb_map_.erase(cb_id);
 }
 
 void AudioDecoderBroker::Reset(base::OnceClosure reset_cb) {
   DVLOG(2) << __func__;
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
 
-  MediaAudioTaskWrapper::CrossThreadOnceResetCB main_loop_cb(
-      WTF::Bind(&AudioDecoderBroker::OnReset, weak_factory_.GetWeakPtr(),
-                std::move(reset_cb)));
+  const int callback_id = CreateCallbackId();
+  pending_reset_cb_map_.insert(callback_id, std::move(reset_cb));
 
   PostCrossThreadTask(
       *media_task_runner_, FROM_HERE,
       WTF::CrossThreadBindOnce(&MediaAudioTaskWrapper::Reset,
                                WTF::CrossThreadUnretained(media_tasks_.get()),
-                               std::move(main_loop_cb)));
+                               callback_id));
 }
 
 bool AudioDecoderBroker::NeedsBitstreamConversion() const {
@@ -352,10 +376,14 @@ bool AudioDecoderBroker::NeedsBitstreamConversion() const {
                           : false;
 }
 
-void AudioDecoderBroker::OnReset(base::OnceClosure reset_cb) {
+void AudioDecoderBroker::OnReset(int cb_id) {
   DVLOG(2) << __func__;
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
-  std::move(reset_cb).Run();
+  DCHECK(pending_reset_cb_map_.Contains(cb_id));
+
+  auto iter = pending_reset_cb_map_.find(cb_id);
+  std::move(iter->value).Run();
+  pending_reset_cb_map_.erase(cb_id);
 }
 
 void AudioDecoderBroker::OnDecodeOutput(

third_party/blink/renderer/modules/webcodecs/audio_decoder_broker.h

@@ -20,6 +20,7 @@
 #include "media/base/decode_status.h"
 #include "third_party/blink/renderer/core/execution_context/execution_context.h"
 #include "third_party/blink/renderer/modules/modules_export.h"
+#include "third_party/blink/renderer/platform/wtf/hash_map.h"
 
 namespace blink {
 
@@ -31,7 +32,20 @@ class MediaAudioTaskWrapper;
 // AudioDecoderBroker, but we need to define it here to implement it below.
 class CrossThreadAudioDecoderClient {
  public:
+  struct DecoderDetails {
+    std::string display_name;
+    bool is_platform_decoder;
+    bool needs_bitstream_conversion;
+  };
+
+  virtual void OnInitialize(media::Status status,
+                            base::Optional<DecoderDetails> details) = 0;
+
+  virtual void OnDecodeDone(int cb_id, media::DecodeStatus status) = 0;
+
   virtual void OnDecodeOutput(scoped_refptr<media::AudioBuffer> buffer) = 0;
+
+  virtual void OnReset(int cb_id) = 0;
 };
 
 // This class brokers the connection between WebCodecs and an underlying
@@ -49,12 +63,6 @@ class MODULES_EXPORT AudioDecoderBroker : public media::AudioDecoder,
  public:
   static constexpr char kDefaultDisplayName[] = "EmptyWebCodecsAudioDecoder";
 
-  struct DecoderDetails {
-    std::string display_name;
-    bool is_platform_decoder;
-    bool needs_bitstream_conversion;
-  };
-
   explicit AudioDecoderBroker(ExecutionContext& execution_context);
   ~AudioDecoderBroker() override;
 
@@ -76,14 +84,26 @@ class MODULES_EXPORT AudioDecoderBroker : public media::AudioDecoder,
   bool NeedsBitstreamConversion() const override;
 
  private:
-  void OnInitialize(InitCB init_cb,
-                    media::Status status,
-                    base::Optional<DecoderDetails> details);
-  void OnDecodeDone(DecodeCB decode_cb, media::DecodeStatus status);
-  void OnReset(base::OnceClosure reset_cb);
+  // Creates a new (incremented) callback ID from |last_callback_id_| for
+  // mapping in |pending_decode_cb_map_|.
+  int CreateCallbackId();
 
   // MediaAudioTaskWrapper::CrossThreadAudioDecoderClient
+  void OnInitialize(media::Status status,
+                    base::Optional<DecoderDetails> details) override;
+  void OnDecodeDone(int cb_id, media::DecodeStatus status) override;
   void OnDecodeOutput(scoped_refptr<media::AudioBuffer> buffer) override;
+  void OnReset(int cb_id) override;
+
+  // Holds the last key for callbacks in the map below. Incremented for each
+  // usage via CreateCallbackId().
+  uint32_t last_callback_id_ = 0;
+
+  // Maps a callback ID to pending Decode(). See CrossThreadVideoDecoderClient.
+  HashMap<int, DecodeCB> pending_decode_cb_map_;
+
+  // Maps a callback ID to pending Reset(). See CrossThreadVideoDecoderClient.
+  HashMap<int, base::OnceClosure> pending_reset_cb_map_;
 
   // Task runner for running codec work (traditionally the media thread).
   scoped_refptr<base::SingleThreadTaskRunner> media_task_runner_;
@@ -94,6 +114,9 @@ class MODULES_EXPORT AudioDecoderBroker : public media::AudioDecoder,
   // Wrapper state for GetDisplayName(), IsPlatformDecoder() and others.
   base::Optional<DecoderDetails> decoder_details_;
 
+  // Pending InitCB saved from the last call to Initialize();
+  InitCB init_cb_;
+
   // OutputCB saved from last call to Initialize().
   OutputCB output_cb_;
 

third_party/blink/renderer/modules/webcodecs/decoder_selector.cc

@@ -94,11 +94,11 @@ void DecoderSelector<StreamType>::SelectDecoder(
   // |impl_| will internally use this the |config| from our NullDemuxerStream.
   demuxer_stream_->Configure(config);
 
-  // Destroying |impl_| will cancel pending operations, so it's safe to use
-  // Unretained() with |select_decoder_cb|.
+  // media::DecoderSelector will call back with a null decoder if selection is
+  // in progress when it is destructed.
   impl_.SelectDecoder(
       WTF::Bind(&DecoderSelector<StreamType>::OnDecoderSelected,
-                WTF::Unretained(this), std::move(select_decoder_cb)),
+                weak_factory_.GetWeakPtr(), std::move(select_decoder_cb)),
       output_cb_);
 }
 

third_party/blink/renderer/modules/webcodecs/decoder_selector.h

@@ -7,6 +7,7 @@
 
 #include <memory>
 
+#include "base/memory/weak_ptr.h"
 #include "media/base/demuxer_stream.h"
 #include "media/base/media_util.h"
 #include "media/filters/decoder_selector.h"
@@ -78,6 +79,8 @@ class DecoderSelector {
 
   // TODO(chcunningham): Route MEDIA_LOG for WebCodecs.
   media::NullMediaLog null_media_log_;
+
+  base::WeakPtrFactory<DecoderSelector<StreamType>> weak_factory_{this};
 };
 
 typedef DecoderSelector<media::DemuxerStream::VIDEO>

third_party/blink/renderer/modules/webcodecs/decoder_template.cc

@@ -4,6 +4,7 @@
 
 #include "third_party/blink/renderer/modules/webcodecs/decoder_template.h"
 
+#include <limits>
 #include <utility>
 #include <vector>
 
@@ -267,9 +268,10 @@ bool DecoderTemplate<Traits>::ProcessDecodeRequest(Request* request) {
 
   // Submit for decoding.
   //
-  // |pending_decode_id_| must not be zero because it is used as a key in a
-  // HeapHashMap (pending_decodes_).
+  // |pending_decode_id_| must not be 0 nor max because it HashMap reserves
+  // these values for "emtpy" and "deleted".
   while (++pending_decode_id_ == 0 ||
+         pending_decode_id_ == std::numeric_limits<uint32_t>::max() ||
          pending_decodes_.Contains(pending_decode_id_))
     ;
   pending_decodes_.Set(pending_decode_id_, request);

third_party/blink/renderer/modules/webcodecs/video_decoder_broker.h

@@ -21,6 +21,7 @@
 #include "media/video/gpu_video_accelerator_factories.h"
 #include "third_party/blink/renderer/core/execution_context/execution_context.h"
 #include "third_party/blink/renderer/modules/modules_export.h"
+#include "third_party/blink/renderer/platform/wtf/hash_map.h"
 
 namespace blink {
 
@@ -30,10 +31,31 @@ class MediaVideoTaskWrapper;
 
 // Client interface for MediaVideoTaskWrapper. Implementation detail of
 // VideoDecoderBroker, but we need to define it here to implement it below.
+//
+// This avoids having pass-through callbacks from main->media task sequence,
+// which is unsafe because the public callers of broker APIs may be broken if
+// their callback is destructed on another thread.
+//
+// An int "cb_id" is used for those that are traditionally OnceCallbacks to
+// lookup the correct public callback.
 class CrossThreadVideoDecoderClient {
  public:
+  struct DecoderDetails {
+    std::string display_name;
+    bool is_platform_decoder;
+    bool needs_bitstream_conversion;
+    int max_decode_requests;
+  };
+
+  virtual void OnInitialize(media::Status status,
+                            base::Optional<DecoderDetails> details) = 0;
+
+  virtual void OnDecodeDone(int cb_id, media::DecodeStatus status) = 0;
+
   virtual void OnDecodeOutput(scoped_refptr<media::VideoFrame> frame,
                               bool can_read_without_stalling) = 0;
+
+  virtual void OnReset(int cb_id) = 0;
 };
 
 // This class brokers the connection between WebCodecs and an underlying
@@ -51,13 +73,6 @@ class MODULES_EXPORT VideoDecoderBroker : public media::VideoDecoder,
  public:
   static constexpr char kDefaultDisplayName[] = "EmptyWebCodecsVideoDecoder";
 
-  struct DecoderDetails {
-    std::string display_name;
-    bool is_platform_decoder;
-    bool needs_bitstream_conversion;
-    int max_decode_requests;
-  };
-
   // |gpu_factories| may be null when GPU accelerated decoding is not available.
   explicit VideoDecoderBroker(
       ExecutionContext& execution_context,
@@ -85,15 +100,17 @@ class MODULES_EXPORT VideoDecoderBroker : public media::VideoDecoder,
   int GetMaxDecodeRequests() const override;
 
  private:
-  void OnInitialize(InitCB init_cb,
-                    media::Status status,
-                    base::Optional<DecoderDetails> details);
-  void OnDecodeDone(DecodeCB decode_cb, media::DecodeStatus status);
-  void OnReset(base::OnceClosure reset_cb);
+  // Creates a new (incremented) callback ID from |last_callback_id_| for
+  // mapping in |pending_decode_cb_map_|.
+  int CreateCallbackId();
 
   // MediaVideoTaskWrapper::CrossThreadVideoDecoderClient
+  void OnInitialize(media::Status status,
+                    base::Optional<DecoderDetails> details) override;
+  void OnDecodeDone(int cb_id, media::DecodeStatus status) override;
   void OnDecodeOutput(scoped_refptr<media::VideoFrame> frame,
                       bool can_read_without_stalling) override;
+  void OnReset(int cb_id) override;
 
   // When media::GpuVideoAcceleratorFactories is provided, its API requires
   // that we use its TaskRunner (the media thread). When not provided, this task
@@ -116,6 +133,19 @@ class MODULES_EXPORT VideoDecoderBroker : public media::VideoDecoder,
   // Set to match the underlying decoder's answer at every OnDecodeOutput().
   bool can_read_without_stalling_ = true;
 
+  // Holds the last key for callbacks in the map below. Incremented for each
+  // usage via CreateCallbackId().
+  uint32_t last_callback_id_ = 0;
+
+  // Maps a callback ID to pending Decode(). See CrossThreadVideoDecoderClient.
+  HashMap<int, DecodeCB> pending_decode_cb_map_;
+
+  // Maps a callback ID to pending Reset(). See CrossThreadVideoDecoderClient.
+  HashMap<int, base::OnceClosure> pending_reset_cb_map_;
+
+  // Pending InitCB saved from the last call to Initialize();
+  InitCB init_cb_;
+
   // OutputCB saved from last call to Initialize().
   OutputCB output_cb_;
 

third_party/blink/web_tests/external/wpt/webcodecs/video-decoder.html

@@ -222,5 +222,38 @@ promise_test(t => {
       () => { throw "flush succeeded unexpectedly"; },
       () => { assert_equals(numErrors, 1, "errors"); });
 }, 'Decode empty VP9 frame');
+
+promise_test(t => {
+  let decoder = new VideoDecoder({
+    output(frame) {
+      t.step(() => {
+        throw "unexpected output";
+      });
+    },
+    error(e) {
+      t.step(() => {
+        // TODO(sandersd): Change to 'throw e' once e is defined.
+        throw "decode error";
+      });
+    }
+  });
+
+  decoder.configure({codec: vp9.codec});
+
+  let fakeChunk = new EncodedVideoChunk('key', 0, Uint8Array.of(0));
+  decoder.decode(fakeChunk);
+
+  // Create the flush promise before closing, as it is invalid to do so later.
+  let flushPromise = decoder.flush();
+
+  // This should synchronously reject the flush() promise.
+  decoder.close();
+
+  // TODO(sandersd): Wait for a bit in case there is a lingering output
+  // or error coming.
+  return flushPromise.then(
+      () => { throw "flush succeeded unexpectedly"; },
+      () => {});
+}, 'Close while decoding corrupt VP9 frame');
 </script>
 </html>