Prevent a race condition in CastAudioOutputStream

- Prevent race condition in CastAudioOutputStream to prevent
FillNextBuffer() from calling AudioSourceCallback::OnMoreData() after
AudioOutputStream::Close() is called.
- FillNextBuffer() will now acquire running_lock_ for the entirety of
its scope.
- CastAudioOutputStream::Close() will synchronously block on
running_lock_ until FillNextBuffer() has completed before setting
running_ to false and continuing its close sequence.
- Subsequent calls to FillNextBuffer() will early return without calling
AudioSourceCallback::OnMoreData().
- Add infrastructure in CastAudioOutputStream unittests to pause and
resume the audio thread.
- Add a test in CastAudioOutputStream unittests to ensure that a call to
Close() will synchronously prevent any further callbacks.

mutex synchronization is removed, race condition is difficult to
reproduce in real-world testing.

Bug: 138695782
Test: Builds, unit tests pass, new unit test passes and fails when the
Change-Id: I11cdd64c0987298ba3fc6a9090359bd2715e6f03
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1761483Reviewed-by: Olga Sharonova <olka@chromium.org>
Reviewed-by: Kenneth MacKay <kmackay@chromium.org>
Reviewed-by: Yuri Wiitala <miu@chromium.org>
Commit-Queue: Hari Nandagopal <hnandagopal@google.com>
Cr-Commit-Position: refs/heads/master@{#691784}

chromecast/media/audio/cast_audio_output_stream.cc

@@ -14,6 +14,7 @@
 #include "base/callback_helpers.h"
 #include "base/location.h"
 #include "base/logging.h"
+#include "base/synchronization/lock.h"
 #include "base/message_loop/message_pump_type.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/threading/thread_task_runner_handle.h"
@@ -113,6 +114,7 @@ class CastAudioOutputStream::CmaWrapper : public CmaBackend::Decoder::Delegate {
              const std::string& device_id,
              CmaBackendFactory* cma_backend_factory);
 
+  void SetRunning(bool running);
   void Initialize(const std::string& application_session_id,
                   chromecast::mojom::MultiroomInfoPtr multiroom_info);
   void Start(AudioSourceCallback* source_callback);
@@ -147,6 +149,8 @@ class CastAudioOutputStream::CmaWrapper : public CmaBackend::Decoder::Delegate {
   const std::string device_id_;
   CmaBackendFactory* const cma_backend_factory_;
 
+  base::Lock running_lock_;
+  bool running_ = true;
   AudioOutputState media_thread_state_;
   CmaBackendState cma_backend_state_ = CmaBackendState::kUinitialized;
   ::media::AudioTimestampHelper timestamp_helper_;
@@ -197,6 +201,11 @@ CastAudioOutputStream::CmaWrapper::CmaWrapper(
   source_callback_ = nullptr;
 }
 
+void CastAudioOutputStream::CmaWrapper::SetRunning(bool running) {
+  base::AutoLock lock(running_lock_);
+  running_ = running;
+}
+
 void CastAudioOutputStream::CmaWrapper::Initialize(
     const std::string& application_session_id,
     chromecast::mojom::MultiroomInfoPtr multiroom_info) {
@@ -355,6 +364,15 @@ void CastAudioOutputStream::CmaWrapper::SetVolume(double volume) {
 void CastAudioOutputStream::CmaWrapper::PushBuffer() {
   DCHECK_CALLED_ON_VALID_THREAD(media_thread_checker_);
 
+  // Acquire running_lock_ for the scope of this push call to
+  // prevent the source callback from closing the output stream
+  // mid-push.
+  base::AutoLock lock(running_lock_);
+
+  // Do not fill more buffers if we have stopped running.
+  if (!running_)
+    return;
+
   // It is possible that this function is called when we are stopped.
   // Return quickly if so.
   if (!source_callback_ || encountered_error_ ||
@@ -480,6 +498,7 @@ class CastAudioOutputStream::MixerServiceWrapper
       MixerServiceConnectionFactory* mixer_service_connection_factory);
   ~MixerServiceWrapper() override = default;
 
+  void SetRunning(bool running);
   void Start(AudioSourceCallback* source_callback);
   void Stop();
   void Close(base::OnceClosure closure);
@@ -507,6 +526,8 @@ class CastAudioOutputStream::MixerServiceWrapper
   std::unique_ptr<media::MixerServiceConnection> mixer_connection_;
   double volume_;
 
+  base::Lock running_lock_;
+  bool running_ = true;
   // MixerServiceWrapper must run on an "io thread".
   base::Thread io_thread_;
   // Task runner on |io_thread_|.
@@ -537,6 +558,11 @@ CastAudioOutputStream::MixerServiceWrapper::MixerServiceWrapper(
   DCHECK(io_task_runner_);
 }
 
+void CastAudioOutputStream::MixerServiceWrapper::SetRunning(bool running) {
+  base::AutoLock lock(running_lock_);
+  running_ = running;
+}
+
 void CastAudioOutputStream::MixerServiceWrapper::Start(
     AudioSourceCallback* source_callback) {
   DCHECK_CALLED_ON_VALID_THREAD(io_thread_checker_);
@@ -600,6 +626,16 @@ void CastAudioOutputStream::MixerServiceWrapper::FillNextBuffer(
     int frames,
     int64_t playout_timestamp) {
   DCHECK_CALLED_ON_VALID_THREAD(io_thread_checker_);
+
+  // Acquire running_lock_ for the scope of this fill call to
+  // prevent the source callback from closing the output stream
+  // mid-fill.
+  base::AutoLock lock(running_lock_);
+
+  // Do not fill more buffers if we have stopped running.
+  if (!running_)
+    return;
+
   if (playout_timestamp < 0) {
     // Assume any negative timestamp is invalid.
     playout_timestamp = 0;
@@ -720,10 +756,16 @@ void CastAudioOutputStream::Close() {
       &CastAudioOutputStream::FinishClose, audio_weak_factory_.GetWeakPtr());
 
   if (mixer_service_wrapper_) {
+    // Synchronously set running to false to guarantee that
+    // AudioSourceCallback::OnMoreData() will not be called anymore.
+    mixer_service_wrapper_->SetRunning(false);
     POST_TO_MIXER_SERVICE_WRAPPER(
         Close, BindToTaskRunner(audio_manager_->GetTaskRunner(),
                                 std::move(finish_callback)));
   } else if (cma_wrapper_) {
+    // Synchronously set running to false to guarantee that
+    // AudioSourceCallback::OnMoreData() will not be called anymore.
+    cma_wrapper_->SetRunning(false);
     POST_TO_CMA_WRAPPER(Close, std::move(finish_callback));
   } else {
     std::move(finish_callback).Run();

chromecast/media/audio/cast_audio_output_stream_unittest.cc

@@ -315,6 +315,34 @@ class CastAudioOutputStreamTest : public ::testing::Test {
     audio_thread_.FlushForTesting();
   }
 
+  static void PauseAndWait(base::WaitableEvent* pause_event,
+                           base::WaitableEvent* resume_event) {
+    pause_event->Signal();
+    resume_event->Wait();
+  }
+
+  // Synchronously pause the audio thread. This function guarantees that
+  // the audio thread will be paused before it returns.
+  void PauseAudioThread() {
+    audio_thread_pause_ = std::make_unique<base::WaitableEvent>(
+        base::WaitableEvent::ResetPolicy::AUTOMATIC,
+        base::WaitableEvent::InitialState::NOT_SIGNALED);
+    audio_thread_resume_ = std::make_unique<base::WaitableEvent>(
+        base::WaitableEvent::ResetPolicy::AUTOMATIC,
+        base::WaitableEvent::InitialState::NOT_SIGNALED);
+    audio_thread_.task_runner()->PostTask(
+        FROM_HERE, base::BindOnce(&PauseAndWait, audio_thread_pause_.get(),
+                                  audio_thread_resume_.get()));
+    audio_thread_pause_->Wait();
+  }
+
+  // Resume a paused audio thread by signalling to it.
+  void ResumeAudioThreadAsync() {
+    if (audio_thread_resume_) {
+      audio_thread_resume_->Signal();
+    }
+  }
+
   ::media::AudioParameters GetAudioParams() {
     return ::media::AudioParameters(format_, channel_layout_, sample_rate_,
                                     frames_per_buffer_);
@@ -332,6 +360,8 @@ class CastAudioOutputStreamTest : public ::testing::Test {
 
   base::Thread audio_thread_;
   base::test::TaskEnvironment task_environment_;
+  std::unique_ptr<base::WaitableEvent> audio_thread_pause_;
+  std::unique_ptr<base::WaitableEvent> audio_thread_resume_;
   std::unique_ptr<MockCmaBackendFactory> mock_backend_factory_;
 
   FakeCmaBackend* cma_backend_ = nullptr;
@@ -540,6 +570,32 @@ TEST_F(CastAudioOutputStreamTest, StopPreventsCallbacks) {
   RunThreadsUntilIdle();
 }
 
+TEST_F(CastAudioOutputStreamTest, ClosePreventsCallbacks) {
+  // Stream API details that Close is synchronous and prevents calls to
+  // callback.
+  ::media::AudioOutputStream* stream = CreateStream();
+  ASSERT_TRUE(stream);
+  ASSERT_TRUE(stream->Open());
+  RunThreadsUntilIdle();
+
+  ::media::MockAudioSourceCallback source_callback;
+  EXPECT_CALL(source_callback, OnMoreData(_, _, _, _))
+      .WillRepeatedly(Invoke(OnMoreData));
+  stream->Start(&source_callback);
+  RunThreadsUntilIdle();
+
+  // Pause the audio thread from running tasks before calling Close() to
+  // prevent it from processing OnMoreData() calls after setting the
+  // expectation and before calling Close().
+  PauseAudioThread();
+  // Once the audio thread resumes work, push/fill calls posted to the audio
+  // thread should no longer call OnMoreData().
+  EXPECT_CALL(source_callback, OnMoreData(_, _, _, _)).Times(0);
+  stream->Close();
+  ResumeAudioThreadAsync();
+  RunThreadsUntilIdle();
+}
+
 TEST_F(CastAudioOutputStreamTest, Format) {
   ::media::AudioParameters::Format format[] = {
       ::media::AudioParameters::AUDIO_PCM_LINEAR,

media/audio/audio_io.h

@@ -110,6 +110,9 @@ class MEDIA_EXPORT AudioOutputStream {
 
   // Close the stream.
   // After calling this method, the object should not be used anymore.
+  // After calling this method, no further AudioSourceCallback methods
+  // should be called on the callback object that was supplied to Start()
+  // by the AudioOutputStream implementation.
   virtual void Close() = 0;
 
   // Flushes the stream. This should only be called if the stream is not