Tab Audio Capture: Browser-side connect/disconnect functionality.

media/audio/audio_output_controller.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "media/audio/audio_output_controller.h"
 
 #include "base/bind.h"
 #include "base/debug/trace_event.h"
 #include "base/message_loop.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/threading/platform_thread.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/time.h"
 #include "build/build_config.h"
 #include "media/audio/shared_memory_util.h"
 
 using base::Time;
 using base::TimeDelta;
 using base::WaitableEvent;
 
 namespace media {
 
 // Polling-related constants.
 const int AudioOutputController::kPollNumAttempts = 3;
 const int AudioOutputController::kPollPauseInMilliseconds = 3;
 
+// An AudioSourceCallback that allows passing around the underlying SyncReader
+// data source between consumers of the data.  It does this in a "pass or
+// delayed pass" scheme that ensures synchronization between timing-sensitive
+// threads is effectively non-blocking.  In addition, SourceSharingCallback
+// ensures that only one consumer is pulling data from the SyncReader at a time.
+class AudioOutputController::SourceSharingCallback
+    : public AudioOutputStream::AudioSourceCallback,
+      public base::RefCountedThreadSafe<SourceSharingCallback> {
+ public:
+  SourceSharingCallback(AudioOutputController* controller, SyncReader* source);
+
+  // Pass the SyncReader to another instance as soon as possible.
+  void PassSoon(const scoped_refptr<SourceSharingCallback>& receiver);
+
+  // AudioSourceCallback implementation.
+  virtual int OnMoreData(AudioBus* dest_bus,
+                         AudioBuffersState buffers_state) OVERRIDE;
+  virtual int OnMoreIOData(AudioBus* source_bus,
+                           AudioBus* dest_bus,
+                           AudioBuffersState buffers_state) OVERRIDE;
+  virtual void OnError(AudioOutputStream* stream, int code) OVERRIDE;
+  virtual void WaitTillDataReady() OVERRIDE;
+
+ private:
+  friend class base::RefCountedThreadSafe<SourceSharingCallback>;
+  virtual ~SourceSharingCallback();
+
+  // Take over reading from |source|.
+  void ReceiveSource(SyncReader* source);
+
+  // Executes a delayed pass (i.e., a pass that was delayed because source_ was
+  // in-use at the time PassSoon() was called).
+  void DoDelayedPassWithLockHeld(
+      const scoped_refptr<SourceSharingCallback>& to);
+
+  // Access to parent/owner instance.
+  AudioOutputController* const controller_;
+
+  // Guards passing of SyncReader to another SourceSharingCallback instance.
+  // The implementation should acquire lock_ for only very short operations
+  // since audio threads are timing-sensitive.
+  base::Lock lock_;
+
+  // Audio data source, or NULL if none yet.
+  SyncReader* source_;
+
+  // True while source_ is in-use without lock_ held.
+  bool is_reading_from_source_;
+
+  // When !delayed_pass_.is_null(), source_ is scheduled to be passed as soon as
+  // possible.
+  base::Closure delayed_pass_;
+
+  DISALLOW_COPY_AND_ASSIGN(SourceSharingCallback);
+};
+
 AudioOutputController::AudioOutputController(AudioManager* audio_manager,
                                              EventHandler* handler,
                                              const AudioParameters& params,
                                              SyncReader* sync_reader)
     : audio_manager_(audio_manager),
+      params_(params),
       handler_(handler),
       stream_(NULL),
+      callback_for_stream_(new SourceSharingCallback(this, sync_reader)),
       volume_(1.0),
       state_(kEmpty),
       sync_reader_(sync_reader),
       message_loop_(audio_manager->GetMessageLoop()),
       number_polling_attempts_left_(0),
-      params_(params),
       ALLOW_THIS_IN_INITIALIZER_LIST(weak_this_(this)) {
 }
 
 AudioOutputController::~AudioOutputController() {
   DCHECK_EQ(kClosed, state_);
 
   if (message_loop_->BelongsToCurrentThread()) {
     DoStopCloseAndClearStream(NULL);
   } else {
     // http://crbug.com/120973
@@ skipping 13 matching lines @@
     AudioManager* audio_manager,
     EventHandler* event_handler,
     const AudioParameters& params,
     SyncReader* sync_reader) {
   DCHECK(audio_manager);
   DCHECK(sync_reader);
 
   if (!params.IsValid() || !audio_manager)
     return NULL;
 
-  // Starts the audio controller thread.
   scoped_refptr<AudioOutputController> controller(new AudioOutputController(
       audio_manager, event_handler, params, sync_reader));
-
   controller->message_loop_->PostTask(FROM_HERE, base::Bind(
       &AudioOutputController::DoCreate, controller));
-
   return controller;
 }
 
 void AudioOutputController::Play() {
   DCHECK(message_loop_);
   message_loop_->PostTask(FROM_HERE, base::Bind(
       &AudioOutputController::DoPlay, this));
 }
 
 void AudioOutputController::Pause() {
@@ skipping 41 matching lines @@
 
   if (!stream_->Open()) {
     state_ = kError;
     DoStopCloseAndClearStream(NULL);
 
     // TODO(hclam): Define error types.
     handler_->OnError(this, 0);
     return;
   }
 
-  // Everything started okay, so register for state change callbacks if we have
-  // not already done so.
-  if (state_ != kRecreating)
-    audio_manager_->AddOutputDeviceChangeListener(this);
+  // Everything started okay, so re-register for state change callbacks.  Note:
+  // The call to DoStopCloseAndClearStream() above called
+  // RemoveOutputDeviceChangeListener().
+  audio_manager_->AddOutputDeviceChangeListener(this);
 
   // We have successfully opened the stream. Set the initial volume.
   stream_->SetVolume(volume_);
 
   // Finally set the state to kCreated.
   State original_state = state_;
   state_ = kCreated;
 
   // And then report we have been created if we haven't done so already.
   if (original_state != kRecreating)
@@ skipping 53 matching lines @@
         weak_this_.GetWeakPtr()),
         TimeDelta::FromMilliseconds(kPollPauseInMilliseconds));
   }
 }
 
 void AudioOutputController::StartStream() {
   DCHECK(message_loop_->BelongsToCurrentThread());
   state_ = kPlaying;
 
   // We start the AudioOutputStream lazily.
-  stream_->Start(this);
+  stream_->Start(callback_for_stream_);
 
   // Tell the event handler that we are now playing.
   handler_->OnPlaying(this);
 }
 
 void AudioOutputController::DoPause() {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
   if (stream_) {
     // Then we stop the audio device. This is not the perfect solution
@@ skipping 26 matching lines @@
 void AudioOutputController::DoFlush() {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
   // TODO(hclam): Actually flush the audio device.
 }
 
 void AudioOutputController::DoClose() {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
   if (state_ != kClosed) {
+    DCHECK(!callback_for_divert_);
     DoStopCloseAndClearStream(NULL);
     sync_reader_->Close();
     state_ = kClosed;
   }
 }
 
 void AudioOutputController::DoSetVolume(double volume) {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
   // Saves the volume to a member first. We may not be able to set the volume
@@ skipping 12 matching lines @@
       return;
   }
 }
 
 void AudioOutputController::DoReportError(int code) {
   DCHECK(message_loop_->BelongsToCurrentThread());
   if (state_ != kClosed)
     handler_->OnError(this, code);
 }
 
-int AudioOutputController::OnMoreData(AudioBus* dest,
-                                      AudioBuffersState buffers_state) {
-  return OnMoreIOData(NULL, dest, buffers_state);
+AudioOutputController::SourceSharingCallback::SourceSharingCallback(
+    AudioOutputController* controller, SyncReader* source)
+    : controller_(controller),
+      source_(source),
+      is_reading_from_source_(false) {
 }
 
-int AudioOutputController::OnMoreIOData(AudioBus* source,
-                                        AudioBus* dest,
-                                        AudioBuffersState buffers_state) {
+AudioOutputController::SourceSharingCallback::~SourceSharingCallback() {
+}
+
+void AudioOutputController::SourceSharingCallback::PassSoon(
+    const scoped_refptr<SourceSharingCallback>& to) {
+  DCHECK_NE(this, to);
+
+  SyncReader* pass_me;
+  {
+    base::AutoLock auto_lock(lock_);
+
+    // Two reasons to delay a pass: 1) another thread is currently reading from
+    // source_; 2) this instance is hasn't received source_ yet.
+    if (is_reading_from_source_ || !source_) {
+      DCHECK(delayed_pass_.is_null()) << "BUG: Attempt to pass twice.";
+      delayed_pass_ = base::Bind(
+          &SourceSharingCallback::DoDelayedPassWithLockHeld, this, to);
+      return;
+    }
+
+    pass_me = source_;
+    source_ = NULL;
+  }
+
+  // Normal path: Pass source_ immediately.
+  to->ReceiveSource(pass_me);
+}
+
+void AudioOutputController::SourceSharingCallback::ReceiveSource(
+    SyncReader* source) {
+  base::AutoLock auto_lock(lock_);
+
+  // Accept the audio data source.
+  DCHECK(!source_)
+      << "BUG: Attempt to receive while already holding a SyncReader.";
+  source_ = source;
+
+  // Execute a delayed pass if PassSoon() scheduled one.
+  if (!delayed_pass_.is_null()) {
+    delayed_pass_.Run();
+    delayed_pass_.Reset();
+  }
+}
+
+void AudioOutputController::SourceSharingCallback::DoDelayedPassWithLockHeld(
+    const scoped_refptr<SourceSharingCallback>& to) {
+  lock_.AssertAcquired();
+
+  // Pass the SyncReader to another SourceSharingCallback instance, but do not
+  // allow the current thread to block on this action.
+  controller_->message_loop_->PostTask(
+      FROM_HERE,
+      base::Bind(&SourceSharingCallback::ReceiveSource, to, source_));
+  source_ = NULL;
+}
+
+int AudioOutputController::SourceSharingCallback::OnMoreData(
+    AudioBus* dest_bus, AudioBuffersState buffers_state) {
+  return OnMoreIOData(NULL, dest_bus, buffers_state);
+}
+
+int AudioOutputController::SourceSharingCallback::OnMoreIOData(
+    AudioBus* source_bus, AudioBus* dest_bus, AudioBuffersState buffers_state) {
   TRACE_EVENT0("audio", "AudioOutputController::OnMoreIOData");
 
+  // Attempt to read from the SyncReader.  While reading, do not hold lock_.
+  // After reading, acquire the lock again; and execute a delayed pass, if
+  // scheduled.
   {
-    // Check state and do nothing if we are not playing.
-    // We are on the hardware audio thread, so lock is needed.
     base::AutoLock auto_lock(lock_);
-    if (state_ != kPlaying) {
-      return 0;
+    if (source_ && controller_->state_ == kPlaying) {
+      is_reading_from_source_ = true;
+      int frames_read;
+      {
+        base::AutoUnlock unlock_while_reading(lock_);
+        frames_read = source_->Read(source_bus, dest_bus);
+        source_->UpdatePendingBytes(
+            buffers_state.total_bytes() +
+            frames_read * controller_->params_.GetBytesPerFrame());
+      }
+      is_reading_from_source_ = false;
+      if (!delayed_pass_.is_null()) {
+        delayed_pass_.Run();
+        delayed_pass_.Reset();
+      }
+      return frames_read;
     }
   }
 
-  int frames = sync_reader_->Read(source, dest);
-  sync_reader_->UpdatePendingBytes(
-      buffers_state.total_bytes() + frames * params_.GetBytesPerFrame());
-  return frames;
+  // We haven't received the SyncReader yet, so fill the destination with zeros.
+  dest_bus->Zero();
+  return dest_bus->frames();
 }
 
-void AudioOutputController::WaitTillDataReady() {
+void AudioOutputController::SourceSharingCallback::WaitTillDataReady() {
 #if defined(OS_WIN) || defined(OS_MACOSX)
   base::Time start = base::Time::Now();
   // Wait for up to 1.5 seconds for DataReady().  1.5 seconds was chosen because
   // it's larger than the playback time of the WaveOut buffer size using the
   // minimum supported sample rate: 4096 / 3000 = ~1.4 seconds.  Even a client
   // expecting real time playout should be able to fill in this time.
   const base::TimeDelta max_wait = base::TimeDelta::FromMilliseconds(1500);
-  while (!sync_reader_->DataReady() &&
+  base::AutoLock auto_lock(lock_);
+  while (source_ && !source_->DataReady() &&
          ((base::Time::Now() - start) < max_wait)) {
+    base::AutoUnlock unlock_while_yielding(lock_);
     base::PlatformThread::YieldCurrentThread();
   }
 #else
   // WaitTillDataReady() is deprecated and should not be used.
   CHECK(false);
 #endif
 }
 
-void AudioOutputController::OnError(AudioOutputStream* stream, int code) {
+void AudioOutputController::SourceSharingCallback::OnError(
+    AudioOutputStream* stream, int code) {
   // Handle error on the audio controller thread.
-  message_loop_->PostTask(FROM_HERE, base::Bind(
-      &AudioOutputController::DoReportError, this, code));
+  controller_->message_loop_->PostTask(FROM_HERE, base::Bind(
+      &AudioOutputController::DoReportError, controller_, code));
 }
 
 void AudioOutputController::DoStopCloseAndClearStream(WaitableEvent* done) {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
   // Allow calling unconditionally and bail if we don't have a stream_ to close.
   if (stream_) {
     stream_->Stop();
     stream_->Close();
     stream_ = NULL;
 
     audio_manager_->RemoveOutputDeviceChangeListener(this);
-    audio_manager_ = NULL;
 
     weak_this_.InvalidateWeakPtrs();
   }
 
   // Should be last in the method, do not touch "this" from here on.
   if (done)
     done->Signal();
 }
 
 void AudioOutputController::OnDeviceChange() {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
   // We should always have a stream by this point.
   CHECK(stream_);
 
-  // Preserve the original state and shutdown the stream.
-  State original_state = state_;
-  stream_->Stop();
-  stream_->Close();
-  stream_ = NULL;
+  // Preserve the original state.
+  const State original_state = state_;
 
-  // Recreate the stream, exit if we ran into an error.
+  // Recreate the stream (DoCreate() will first shut down an existing stream).
+  // Exit if we ran into an error.
   state_ = kRecreating;
   DoCreate();
   if (!stream_ || state_ == kError)
     return;
 
   // Get us back to the original state or an equivalent state.
   switch (original_state) {
     case kStarting:
     case kPlaying:
+      DoSetVolume(volume_);
       DoPlay();
       return;
     case kCreated:
     case kPausedWhenStarting:
     case kPaused:
       // From the outside these three states are equivalent.
       return;
     default:
       NOTREACHED() << "Invalid original state.";
   }
 }
 
+AudioOutputStream::AudioSourceCallback* AudioOutputController::Divert() {
+  DCHECK(!callback_for_divert_) << "BUG: Already diverted!";
+  callback_for_divert_ = new SourceSharingCallback(this, NULL);
+  callback_for_stream_->PassSoon(callback_for_divert_);
+  return callback_for_divert_;
+}
+
+void AudioOutputController::Revert(
+    AudioOutputStream::AudioSourceCallback* asc) {
+  DCHECK_EQ(callback_for_divert_, asc);
+  callback_for_divert_->PassSoon(callback_for_stream_);
+  callback_for_divert_ = NULL;
+}
+
 }  // namespace media