Fix race between enabled() and set_enabled() in VideoTrack.

Along the way I introduced VideoSourceBaseGuarded, which is equivalent
to VideoSourceBase except that it applies thread checks. I found that
it's easy to use VideoSourceBase incorrectly and in fact there appear
to be tests that do this.

I made the source object const in VideoTrack, as it already was in
AudioTrack, and that allowed for making the GetSource() accessors
bypass the proxy thread hop and give the caller direct access.

Bug: webrtc:12773, b/188139639, webrtc:12780
Change-Id: I022175c4239a1306ef54059c131d81411d5124fe
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/219160
Reviewed-by: Mirko Bonadei <mbonadei@webrtc.org>
Reviewed-by: Andrey Logvin <landrey@webrtc.org>
Commit-Queue: Tommi <tommi@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#34096}

api/media_stream_track_proxy.h

@@ -30,7 +30,7 @@ BYPASS_PROXY_CONSTMETHOD0(std::string, kind)
 BYPASS_PROXY_CONSTMETHOD0(std::string, id)
 PROXY_CONSTMETHOD0(TrackState, state)
 PROXY_CONSTMETHOD0(bool, enabled)
-PROXY_CONSTMETHOD0(AudioSourceInterface*, GetSource)
+BYPASS_PROXY_CONSTMETHOD0(AudioSourceInterface*, GetSource)
 PROXY_METHOD1(void, AddSink, AudioTrackSinkInterface*)
 PROXY_METHOD1(void, RemoveSink, AudioTrackSinkInterface*)
 PROXY_METHOD1(bool, GetSignalLevel, int*)
@@ -44,17 +44,17 @@ BEGIN_PROXY_MAP(VideoTrack)
 PROXY_PRIMARY_THREAD_DESTRUCTOR()
 BYPASS_PROXY_CONSTMETHOD0(std::string, kind)
 BYPASS_PROXY_CONSTMETHOD0(std::string, id)
-PROXY_CONSTMETHOD0(TrackState, state)
-PROXY_CONSTMETHOD0(bool, enabled)
-PROXY_METHOD1(bool, set_enabled, bool)
-PROXY_CONSTMETHOD0(ContentHint, content_hint)
-PROXY_METHOD1(void, set_content_hint, ContentHint)
+PROXY_SECONDARY_CONSTMETHOD0(TrackState, state)
+PROXY_SECONDARY_CONSTMETHOD0(bool, enabled)
+PROXY_SECONDARY_METHOD1(bool, set_enabled, bool)
+PROXY_SECONDARY_CONSTMETHOD0(ContentHint, content_hint)
+PROXY_SECONDARY_METHOD1(void, set_content_hint, ContentHint)
 PROXY_SECONDARY_METHOD2(void,
                         AddOrUpdateSink,
                         rtc::VideoSinkInterface<VideoFrame>*,
                         const rtc::VideoSinkWants&)
 PROXY_SECONDARY_METHOD1(void, RemoveSink, rtc::VideoSinkInterface<VideoFrame>*)
-PROXY_CONSTMETHOD0(VideoTrackSourceInterface*, GetSource)
+BYPASS_PROXY_CONSTMETHOD0(VideoTrackSourceInterface*, GetSource)
 
 PROXY_METHOD1(void, RegisterObserver, ObserverInterface*)
 PROXY_METHOD1(void, UnregisterObserver, ObserverInterface*)

api/proxy.h

@@ -435,10 +435,6 @@ class ConstMethodCall : public QueuedTask {
 #define BYPASS_PROXY_CONSTMETHOD0(r, method)                                \
   r method() const override {                                               \
     proxy_internal::TraceApiCall(class_name_, PROXY_STRINGIZE(method));     \
-    static_assert(                                                          \
-        std::is_same<r, rtc::Thread*>::value || !std::is_pointer<r>::value, \
-        "Type is a pointer");                                               \
-    static_assert(!std::is_reference<r>::value, "Type is a reference");     \
     return c_->method();                                                    \
   }
 

media/BUILD.gn

@@ -106,6 +106,7 @@ rtc_library("rtc_media_base") {
     "../rtc_base:stringutils",
     "../rtc_base/synchronization:mutex",
     "../rtc_base/system:file_wrapper",
+    "../rtc_base/system:no_unique_address",
     "../rtc_base/system:rtc_export",
     "../rtc_base/task_utils:pending_task_safety_flag",
     "../rtc_base/task_utils:to_queued_task",

media/base/video_source_base.cc

@@ -10,6 +10,8 @@
 
 #include "media/base/video_source_base.h"
 
+#include <algorithm>
+
 #include "absl/algorithm/container.h"
 #include "rtc_base/checks.h"
 
@@ -52,4 +54,51 @@ VideoSourceBase::SinkPair* VideoSourceBase::FindSinkPair(
   return nullptr;
 }
 
+VideoSourceBaseGuarded::VideoSourceBaseGuarded() = default;
+VideoSourceBaseGuarded::~VideoSourceBaseGuarded() = default;
+
+void VideoSourceBaseGuarded::AddOrUpdateSink(
+    VideoSinkInterface<webrtc::VideoFrame>* sink,
+    const VideoSinkWants& wants) {
+  RTC_DCHECK_RUN_ON(&source_sequence_);
+  RTC_DCHECK(sink != nullptr);
+
+  SinkPair* sink_pair = FindSinkPair(sink);
+  if (!sink_pair) {
+    sinks_.push_back(SinkPair(sink, wants));
+  } else {
+    sink_pair->wants = wants;
+  }
+}
+
+void VideoSourceBaseGuarded::RemoveSink(
+    VideoSinkInterface<webrtc::VideoFrame>* sink) {
+  RTC_DCHECK_RUN_ON(&source_sequence_);
+  RTC_DCHECK(sink != nullptr);
+  RTC_DCHECK(FindSinkPair(sink));
+  sinks_.erase(std::remove_if(sinks_.begin(), sinks_.end(),
+                              [sink](const SinkPair& sink_pair) {
+                                return sink_pair.sink == sink;
+                              }),
+               sinks_.end());
+}
+
+VideoSourceBaseGuarded::SinkPair* VideoSourceBaseGuarded::FindSinkPair(
+    const VideoSinkInterface<webrtc::VideoFrame>* sink) {
+  RTC_DCHECK_RUN_ON(&source_sequence_);
+  auto sink_pair_it = absl::c_find_if(
+      sinks_,
+      [sink](const SinkPair& sink_pair) { return sink_pair.sink == sink; });
+  if (sink_pair_it != sinks_.end()) {
+    return &*sink_pair_it;
+  }
+  return nullptr;
+}
+
+const std::vector<VideoSourceBaseGuarded::SinkPair>&
+VideoSourceBaseGuarded::sink_pairs() const {
+  RTC_DCHECK_RUN_ON(&source_sequence_);
+  return sinks_;
+}
+
 }  // namespace rtc

media/base/video_source_base.h

@@ -17,10 +17,14 @@
 #include "api/video/video_frame.h"
 #include "api/video/video_sink_interface.h"
 #include "api/video/video_source_interface.h"
+#include "rtc_base/system/no_unique_address.h"
 
 namespace rtc {
 
-// VideoSourceBase is not thread safe.
+// VideoSourceBase is not thread safe. Before using this class, consider using
+// VideoSourceBaseGuarded below instead, which is an identical implementation
+// but applies a sequence checker to help protect internal state.
+// TODO(bugs.webrtc.org/12780): Delete this class.
 class VideoSourceBase : public VideoSourceInterface<webrtc::VideoFrame> {
  public:
   VideoSourceBase();
@@ -44,6 +48,36 @@ class VideoSourceBase : public VideoSourceInterface<webrtc::VideoFrame> {
   std::vector<SinkPair> sinks_;
 };
 
+// VideoSourceBaseGuarded assumes that operations related to sinks, occur on the
+// same TQ/thread that the object was constructed on.
+class VideoSourceBaseGuarded : public VideoSourceInterface<webrtc::VideoFrame> {
+ public:
+  VideoSourceBaseGuarded();
+  ~VideoSourceBaseGuarded() override;
+
+  void AddOrUpdateSink(VideoSinkInterface<webrtc::VideoFrame>* sink,
+                       const VideoSinkWants& wants) override;
+  void RemoveSink(VideoSinkInterface<webrtc::VideoFrame>* sink) override;
+
+ protected:
+  struct SinkPair {
+    SinkPair(VideoSinkInterface<webrtc::VideoFrame>* sink, VideoSinkWants wants)
+        : sink(sink), wants(wants) {}
+    VideoSinkInterface<webrtc::VideoFrame>* sink;
+    VideoSinkWants wants;
+  };
+
+  SinkPair* FindSinkPair(const VideoSinkInterface<webrtc::VideoFrame>* sink);
+  const std::vector<SinkPair>& sink_pairs() const;
+
+  // Keep the `source_sequence_` checker protected to allow sub classes the
+  // ability to call Detach() if/when appropriate.
+  RTC_NO_UNIQUE_ADDRESS webrtc::SequenceChecker source_sequence_;
+
+ private:
+  std::vector<SinkPair> sinks_ RTC_GUARDED_BY(&source_sequence_);
+};
+
 }  // namespace rtc
 
 #endif  // MEDIA_BASE_VIDEO_SOURCE_BASE_H_

pc/audio_track.cc

@@ -32,7 +32,7 @@ AudioTrack::AudioTrack(const std::string& label,
 }
 
 AudioTrack::~AudioTrack() {
-  RTC_DCHECK(thread_checker_.IsCurrent());
+  RTC_DCHECK_RUN_ON(&thread_checker_);
   set_state(MediaStreamTrackInterface::kEnded);
   if (audio_source_)
     audio_source_->UnregisterObserver(this);
@@ -43,24 +43,24 @@ std::string AudioTrack::kind() const {
 }
 
 AudioSourceInterface* AudioTrack::GetSource() const {
-  RTC_DCHECK(thread_checker_.IsCurrent());
+  RTC_DCHECK_RUN_ON(&thread_checker_);
   return audio_source_.get();
 }
 
 void AudioTrack::AddSink(AudioTrackSinkInterface* sink) {
-  RTC_DCHECK(thread_checker_.IsCurrent());
+  RTC_DCHECK_RUN_ON(&thread_checker_);
   if (audio_source_)
     audio_source_->AddSink(sink);
 }
 
 void AudioTrack::RemoveSink(AudioTrackSinkInterface* sink) {
-  RTC_DCHECK(thread_checker_.IsCurrent());
+  RTC_DCHECK_RUN_ON(&thread_checker_);
   if (audio_source_)
     audio_source_->RemoveSink(sink);
 }
 
 void AudioTrack::OnChanged() {
-  RTC_DCHECK(thread_checker_.IsCurrent());
+  RTC_DCHECK_RUN_ON(&thread_checker_);
   if (audio_source_->state() == MediaSourceInterface::kEnded) {
     set_state(kEnded);
   } else {

pc/audio_track.h

@@ -41,13 +41,13 @@ class AudioTrack : public MediaStreamTrack<AudioTrackInterface>,
   // MediaStreamTrack implementation.
   std::string kind() const override;
 
- private:
   // AudioTrackInterface implementation.
   AudioSourceInterface* GetSource() const override;
 
   void AddSink(AudioTrackSinkInterface* sink) override;
   void RemoveSink(AudioTrackSinkInterface* sink) override;
 
+ private:
   // ObserverInterface implementation.
   void OnChanged() override;
 

pc/track_media_info_map_unittest.cc

@@ -31,6 +31,45 @@ namespace webrtc {
 
 namespace {
 
+class MockVideoTrack : public VideoTrackInterface {
+ public:
+  // NotifierInterface
+  MOCK_METHOD(void,
+              RegisterObserver,
+              (ObserverInterface * observer),
+              (override));
+  MOCK_METHOD(void,
+              UnregisterObserver,
+              (ObserverInterface * observer),
+              (override));
+
+  // MediaStreamTrackInterface
+  MOCK_METHOD(std::string, kind, (), (const, override));
+  MOCK_METHOD(std::string, id, (), (const, override));
+  MOCK_METHOD(bool, enabled, (), (const, override));
+  MOCK_METHOD(bool, set_enabled, (bool enable), (override));
+  MOCK_METHOD(TrackState, state, (), (const, override));
+
+  // VideoSourceInterface
+  MOCK_METHOD(void,
+              AddOrUpdateSink,
+              (rtc::VideoSinkInterface<VideoFrame> * sink,
+               const rtc::VideoSinkWants& wants),
+              (override));
+  // RemoveSink must guarantee that at the time the method returns,
+  // there is no current and no future calls to VideoSinkInterface::OnFrame.
+  MOCK_METHOD(void,
+              RemoveSink,
+              (rtc::VideoSinkInterface<VideoFrame> * sink),
+              (override));
+
+  // VideoTrackInterface
+  MOCK_METHOD(VideoTrackSourceInterface*, GetSource, (), (const, override));
+
+  MOCK_METHOD(ContentHint, content_hint, (), (const, override));
+  MOCK_METHOD(void, set_content_hint, (ContentHint hint), (override));
+};
+
 RtpParameters CreateRtpParametersWithSsrcs(
     std::initializer_list<uint32_t> ssrcs) {
   RtpParameters params;
@@ -79,23 +118,35 @@ rtc::scoped_refptr<MockRtpReceiverInternal> CreateMockRtpReceiver(
   return receiver;
 }
 
+rtc::scoped_refptr<VideoTrackInterface> CreateVideoTrack(
+    const std::string& id) {
+  return VideoTrack::Create(id, FakeVideoTrackSource::Create(false),
+                            rtc::Thread::Current());
+}
+
+rtc::scoped_refptr<VideoTrackInterface> CreateMockVideoTrack(
+    const std::string& id) {
+  auto track = rtc::make_ref_counted<MockVideoTrack>();
+  EXPECT_CALL(*track, kind())
+      .WillRepeatedly(::testing::Return(VideoTrack::kVideoKind));
+  return track;
+}
+
 class TrackMediaInfoMapTest : public ::testing::Test {
  public:
   TrackMediaInfoMapTest() : TrackMediaInfoMapTest(true) {}
 
-  explicit TrackMediaInfoMapTest(bool use_current_thread)
+  explicit TrackMediaInfoMapTest(bool use_real_video_track)
       : voice_media_info_(new cricket::VoiceMediaInfo()),
         video_media_info_(new cricket::VideoMediaInfo()),
         local_audio_track_(AudioTrack::Create("LocalAudioTrack", nullptr)),
         remote_audio_track_(AudioTrack::Create("RemoteAudioTrack", nullptr)),
-        local_video_track_(VideoTrack::Create(
-            "LocalVideoTrack",
-            FakeVideoTrackSource::Create(false),
-            use_current_thread ? rtc::Thread::Current() : nullptr)),
-        remote_video_track_(VideoTrack::Create(
-            "RemoteVideoTrack",
-            FakeVideoTrackSource::Create(false),
-            use_current_thread ? rtc::Thread::Current() : nullptr)) {}
+        local_video_track_(use_real_video_track
+                               ? CreateVideoTrack("LocalVideoTrack")
+                               : CreateMockVideoTrack("LocalVideoTrack")),
+        remote_video_track_(use_real_video_track
+                                ? CreateVideoTrack("RemoteVideoTrack")
+                                : CreateMockVideoTrack("LocalVideoTrack")) {}
 
   ~TrackMediaInfoMapTest() {
     // If we have a map the ownership has been passed to the map, only delete if
@@ -179,8 +230,8 @@ class TrackMediaInfoMapTest : public ::testing::Test {
   std::unique_ptr<TrackMediaInfoMap> map_;
   rtc::scoped_refptr<AudioTrack> local_audio_track_;
   rtc::scoped_refptr<AudioTrack> remote_audio_track_;
-  rtc::scoped_refptr<VideoTrack> local_video_track_;
-  rtc::scoped_refptr<VideoTrack> remote_video_track_;
+  rtc::scoped_refptr<VideoTrackInterface> local_video_track_;
+  rtc::scoped_refptr<VideoTrackInterface> remote_video_track_;
 };
 
 }  // namespace

pc/video_rtp_receiver.cc

@@ -49,7 +49,7 @@ VideoRtpReceiver::VideoRtpReceiver(
       attachment_id_(GenerateUniqueId()) {
   RTC_DCHECK(worker_thread_);
   SetStreams(streams);
-  source_->SetState(MediaSourceInterface::kLive);
+  RTC_DCHECK_EQ(source_->state(), MediaSourceInterface::kLive);
 }
 
 VideoRtpReceiver::~VideoRtpReceiver() {

pc/video_track.cc

@@ -11,6 +11,7 @@
 #include "pc/video_track.h"
 
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "api/notifier.h"
@@ -28,10 +29,16 @@ VideoTrack::VideoTrack(const std::string& label,
       worker_thread_(worker_thread),
       video_source_(video_source),
       content_hint_(ContentHint::kNone) {
+  RTC_DCHECK_RUN_ON(&signaling_thread_);
+  // Detach the thread checker for VideoSourceBaseGuarded since we'll make calls
+  // to VideoSourceBaseGuarded on the worker thread, but we're currently on the
+  // signaling thread.
+  source_sequence_.Detach();
   video_source_->RegisterObserver(this);
 }
 
 VideoTrack::~VideoTrack() {
+  RTC_DCHECK_RUN_ON(&signaling_thread_);
   video_source_->UnregisterObserver(this);
 }
 
@@ -43,26 +50,31 @@ std::string VideoTrack::kind() const {
 // thread.
 void VideoTrack::AddOrUpdateSink(rtc::VideoSinkInterface<VideoFrame>* sink,
                                  const rtc::VideoSinkWants& wants) {
-  RTC_DCHECK(worker_thread_->IsCurrent());
-  VideoSourceBase::AddOrUpdateSink(sink, wants);
+  RTC_DCHECK_RUN_ON(worker_thread_);
+  VideoSourceBaseGuarded::AddOrUpdateSink(sink, wants);
   rtc::VideoSinkWants modified_wants = wants;
   modified_wants.black_frames = !enabled();
   video_source_->AddOrUpdateSink(sink, modified_wants);
 }
 
 void VideoTrack::RemoveSink(rtc::VideoSinkInterface<VideoFrame>* sink) {
-  RTC_DCHECK(worker_thread_->IsCurrent());
-  VideoSourceBase::RemoveSink(sink);
+  RTC_DCHECK_RUN_ON(worker_thread_);
+  VideoSourceBaseGuarded::RemoveSink(sink);
   video_source_->RemoveSink(sink);
 }
 
+VideoTrackSourceInterface* VideoTrack::GetSource() const {
+  RTC_DCHECK_RUN_ON(&signaling_thread_);
+  return video_source_.get();
+}
+
 VideoTrackInterface::ContentHint VideoTrack::content_hint() const {
-  RTC_DCHECK_RUN_ON(&signaling_thread_checker_);
+  RTC_DCHECK_RUN_ON(worker_thread_);
   return content_hint_;
 }
 
 void VideoTrack::set_content_hint(ContentHint hint) {
-  RTC_DCHECK_RUN_ON(&signaling_thread_checker_);
+  RTC_DCHECK_RUN_ON(worker_thread_);
   if (content_hint_ == hint)
     return;
   content_hint_ = hint;
@@ -70,25 +82,36 @@ void VideoTrack::set_content_hint(ContentHint hint) {
 }
 
 bool VideoTrack::set_enabled(bool enable) {
-  RTC_DCHECK(signaling_thread_checker_.IsCurrent());
-  worker_thread_->Invoke<void>(RTC_FROM_HERE, [enable, this] {
-    RTC_DCHECK(worker_thread_->IsCurrent());
-    for (auto& sink_pair : sink_pairs()) {
-      rtc::VideoSinkWants modified_wants = sink_pair.wants;
-      modified_wants.black_frames = !enable;
-      video_source_->AddOrUpdateSink(sink_pair.sink, modified_wants);
-    }
-  });
+  RTC_DCHECK_RUN_ON(worker_thread_);
+  for (auto& sink_pair : sink_pairs()) {
+    rtc::VideoSinkWants modified_wants = sink_pair.wants;
+    modified_wants.black_frames = !enable;
+    video_source_->AddOrUpdateSink(sink_pair.sink, modified_wants);
+  }
   return MediaStreamTrack<VideoTrackInterface>::set_enabled(enable);
 }
 
+bool VideoTrack::enabled() const {
+  RTC_DCHECK_RUN_ON(worker_thread_);
+  return MediaStreamTrack<VideoTrackInterface>::enabled();
+}
+
+MediaStreamTrackInterface::TrackState VideoTrack::state() const {
+  RTC_DCHECK_RUN_ON(worker_thread_);
+  return MediaStreamTrack<VideoTrackInterface>::state();
+}
+
 void VideoTrack::OnChanged() {
-  RTC_DCHECK(signaling_thread_checker_.IsCurrent());
-  if (video_source_->state() == MediaSourceInterface::kEnded) {
-    set_state(kEnded);
-  } else {
-    set_state(kLive);
-  }
+  RTC_DCHECK_RUN_ON(&signaling_thread_);
+  worker_thread_->Invoke<void>(
+      RTC_FROM_HERE, [this, state = video_source_->state()]() {
+        // TODO(tommi): Calling set_state() this way isn't ideal since we're
+        // currently blocking the signaling thread and set_state() may
+        // internally fire notifications via `FireOnChanged()` which may further
+        // amplify the blocking effect on the signaling thread.
+        rtc::Thread::ScopedDisallowBlockingCalls no_blocking_calls;
+        set_state(state == MediaSourceInterface::kEnded ? kEnded : kLive);
+      });
 }
 
 rtc::scoped_refptr<VideoTrack> VideoTrack::Create(

pc/video_track.h

@@ -27,7 +27,7 @@
 namespace webrtc {
 
 class VideoTrack : public MediaStreamTrack<VideoTrackInterface>,
-                   public rtc::VideoSourceBase,
+                   public rtc::VideoSourceBaseGuarded,
                    public ObserverInterface {
  public:
   static rtc::scoped_refptr<VideoTrack> Create(
@@ -38,13 +38,13 @@ class VideoTrack : public MediaStreamTrack<VideoTrackInterface>,
   void AddOrUpdateSink(rtc::VideoSinkInterface<VideoFrame>* sink,
                        const rtc::VideoSinkWants& wants) override;
   void RemoveSink(rtc::VideoSinkInterface<VideoFrame>* sink) override;
+  VideoTrackSourceInterface* GetSource() const override;
 
-  VideoTrackSourceInterface* GetSource() const override {
-    return video_source_.get();
-  }
   ContentHint content_hint() const override;
   void set_content_hint(ContentHint hint) override;
   bool set_enabled(bool enable) override;
+  bool enabled() const override;
+  MediaStreamTrackInterface::TrackState state() const override;
   std::string kind() const override;
 
  protected:
@@ -57,10 +57,10 @@ class VideoTrack : public MediaStreamTrack<VideoTrackInterface>,
   // Implements ObserverInterface. Observes |video_source_| state.
   void OnChanged() override;
 
+  RTC_NO_UNIQUE_ADDRESS webrtc::SequenceChecker signaling_thread_;
   rtc::Thread* const worker_thread_;
-  SequenceChecker signaling_thread_checker_;
-  rtc::scoped_refptr<VideoTrackSourceInterface> video_source_;
-  ContentHint content_hint_ RTC_GUARDED_BY(signaling_thread_checker_);
+  const rtc::scoped_refptr<VideoTrackSourceInterface> video_source_;
+  ContentHint content_hint_ RTC_GUARDED_BY(worker_thread_);
 };
 
 }  // namespace webrtc

pc/video_track_source.cc

@@ -15,7 +15,7 @@
 namespace webrtc {
 
 VideoTrackSource::VideoTrackSource(bool remote)
-    : state_(kInitializing), remote_(remote) {
+    : state_(kLive), remote_(remote) {
   worker_thread_checker_.Detach();
 }