Threadsafe access to AudioSource callback. Release stream on close. Handle error on open better.

This should fix the memory leak on stream close by synchronizing the removal
of the audio source callback with the signaling of a close event to both the
audio source, and to the audio manager.

This change also makes it possible to go from kPlaying -> kPlaying because
the audio source callback can be swapped correctly.

Lastly, Open() is made more robust by failing early on an error state.

BUG=19860,18217
TEST=none

Review URL: http://codereview.chromium.org/173168

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@23988 0039d316-1c4b-4281-b951-d872f2087c98

5 files changed, 151 insertions, 46 deletions

diff --git a/media/audio/linux/alsa_output.cc b/media/audio/linux/alsa_output.cc
index d0fda06..522b8c4 100644
--- a/media/audio/linux/alsa_output.cc
+++ b/media/audio/linux/alsa_output.cc
@@ -34,10 +34,11 @@
 // This reduces the need for critical sections because the public API code can
 // assume that no mutations occur to the |shared_data_| between queries.
 //
-// On the message loop side, most tasks have been coded such that they can
+// On the message loop side, tasks have been coded such that they can
 // operate safely regardless of when state changes happen to |shared_data_|.
-// Code that is sensitive to the timing holds the |shared_data_.lock_|
-// explicitly for the duration of the critical section.
+// Code that is sensitive to the timing of state changes are delegated to the
+// |shared_data_| object so they can executed while holding
+// |shared_data_.lock_|.
 //
 //
 // SEMANTICS OF CloseTask()
@@ -81,6 +82,7 @@
 #include "base/time.h"
 #include "media/audio/audio_util.h"
 #include "media/audio/linux/alsa_wrapper.h"
+#include "media/audio/linux/audio_manager_linux.h"
 
 // Amount of time to wait if we've exhausted the data source.  This is to avoid
 // busy looping.
@@ -150,6 +152,7 @@ AlsaPcmOutputStream::AlsaPcmOutputStream(const std::string& device_name,
                                          int sample_rate,
                                          int bits_per_sample,
                                          AlsaWrapper* wrapper,
+                                         AudioManagerLinux* manager,
                                          MessageLoop* message_loop)
     : shared_data_(MessageLoop::current()),
       device_name_(device_name),
@@ -160,8 +163,8 @@ AlsaPcmOutputStream::AlsaPcmOutputStream(const std::string& device_name,
       bytes_per_frame_(channels_ * bits_per_sample / 8),
       stop_stream_(false),
       wrapper_(wrapper),
+      manager_(manager),
       playback_handle_(NULL),
-      source_callback_(NULL),
       frames_per_packet_(0),
       client_thread_loop_(MessageLoop::current()),
       message_loop_(message_loop) {
@@ -205,6 +208,10 @@ bool AlsaPcmOutputStream::Open(size_t packet_size) {
       << "Buffers should end on a frame boundary. Frame size: "
       << bytes_per_frame_;
 
+  if (shared_data_.state() == kInError) {
+    return false;
+  }
+
   if (!shared_data_.CanTransitionTo(kIsOpened)) {
     NOTREACHED() << "Invalid state: " << shared_data_.state();
     return false;
@@ -256,11 +263,24 @@ bool AlsaPcmOutputStream::Open(size_t packet_size) {
 void AlsaPcmOutputStream::Close() {
   DCHECK_EQ(MessageLoop::current(), client_thread_loop_);
 
-  if (shared_data_.TransitionTo(kIsClosed) == kIsClosed) {
-    message_loop_->PostTask(
-        FROM_HERE,
-        NewRunnableMethod(this, &AlsaPcmOutputStream::CloseTask));
+  // Sanity check that the transition occurs correctly.  It is safe to
+  // continue anyways because all operations for closing are idempotent.
+  if (shared_data_.TransitionTo(kIsClosed) != kIsClosed) {
+    NOTREACHED() << "Unable to transition Closed.";
   }
+
+  // Signal our successful close, and disassociate the source callback.
+  shared_data_.OnClose(this);
+  shared_data_.set_source_callback(NULL);
+
+  message_loop_->PostTask(
+      FROM_HERE,
+      NewRunnableMethod(this, &AlsaPcmOutputStream::CloseTask));
+
+  // Signal to the manager that we're closed and can be removed.  Since
+  // we just posted a CloseTask to the message loop, we won't be deleted
+  // immediately, but it will happen soon afterwards.
+  manager()->ReleaseStream(this);
 }
 
 void AlsaPcmOutputStream::Start(AudioSourceCallback* callback) {
@@ -268,11 +288,13 @@ void AlsaPcmOutputStream::Start(AudioSourceCallback* callback) {
 
   CHECK(callback);
 
+  shared_data_.set_source_callback(callback);
+
   // Only post the task if we can enter the playing state.
   if (shared_data_.TransitionTo(kIsPlaying) == kIsPlaying) {
     message_loop_->PostTask(
         FROM_HERE,
-        NewRunnableMethod(this, &AlsaPcmOutputStream::StartTask, callback));
+        NewRunnableMethod(this, &AlsaPcmOutputStream::StartTask));
   }
 }
 
@@ -303,11 +325,9 @@ void AlsaPcmOutputStream::FinishOpen(snd_pcm_t* playback_handle,
   frames_per_packet_ = packet_size / bytes_per_frame_;
 }
 
-void AlsaPcmOutputStream::StartTask(AudioSourceCallback* callback) {
+void AlsaPcmOutputStream::StartTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
-  source_callback_ = callback;
-
   // When starting again, drop all packets in the device and prepare it again
   // incase we are restarting from a pause state and need to flush old data.
   int error = wrapper_->PcmDrop(playback_handle_);
@@ -341,6 +361,8 @@ void AlsaPcmOutputStream::StartTask(AudioSourceCallback* callback) {
 }
 
 void AlsaPcmOutputStream::CloseTask() {
+  // NOTE: Keep this function idempotent to handle errors that might cause
+  // multiple CloseTasks to be posted.
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   // Shutdown the audio device.
@@ -352,15 +374,6 @@ void AlsaPcmOutputStream::CloseTask() {
   // Release the buffer.
   packet_.reset();
 
-  // The |source_callback_| may be NULL if the stream is being closed before it
-  // was ever started.
-  if (source_callback_) {
-    // TODO(ajwong): We need to call source_callback_->OnClose(), but the
-    // ownerships of the callback is broken right now, so we'd crash.  Instead,
-    // just leak. Bug 18217.
-    source_callback_ = NULL;
-  }
-
   // Signal anything that might already be scheduled to stop.
   stop_stream_ = true;
 }
@@ -377,8 +390,8 @@ void AlsaPcmOutputStream::BufferPacket(Packet* packet) {
   // Request more data if we don't have any cached.
   if (packet->used >= packet->size) {
     packet->used = 0;
-    packet->size = source_callback_->OnMoreData(this, packet->buffer.get(),
-                                                packet->capacity);
+    packet->size = shared_data_.OnMoreData(this, packet->buffer.get(),
+                                           packet->capacity);
     CHECK(packet->size <= packet->capacity) << "Data source overran buffer.";
 
     // This should not happen, but incase it does, drop any trailing bytes
@@ -431,9 +444,7 @@ void AlsaPcmOutputStream::WritePacket(Packet* packet) {
       if (frames_written != -EAGAIN) {
         LOG(ERROR) << "Failed to write to pcm device: "
                    << wrapper_->StrError(frames_written);
-        // TODO(ajwong): We need to call source_callback_->OnError(), but the
-        // ownerships of the callback is broken right now, so we'd crash.
-        // Instead, just leak.  Bug 18217.
+        shared_data_.OnError(this, frames_written);
         stop_stream_ = true;
       }
     } else {
@@ -540,10 +551,16 @@ snd_pcm_sframes_t AlsaPcmOutputStream::GetAvailableFrames() {
   return available_frames;
 }
 
+AudioManagerLinux* AlsaPcmOutputStream::manager() {
+  DCHECK_EQ(MessageLoop::current(), client_thread_loop_);
+  return manager_;
+}
+
 AlsaPcmOutputStream::SharedData::SharedData(
     MessageLoop* state_transition_loop)
     : state_(kCreated),
       volume_(1.0f),
+      source_callback_(NULL),
       state_transition_loop_(state_transition_loop) {
 }
 
@@ -565,7 +582,8 @@ bool AlsaPcmOutputStream::SharedData::CanTransitionTo_Locked(
           to == kIsClosed || to == kInError;
 
     case kIsPlaying:
-      return to == kIsStopped || to == kIsClosed || to == kInError;
+      return to == kIsPlaying || to == kIsStopped ||
+          to == kIsClosed || to == kInError;
 
     case kIsStopped:
       return to == kIsPlaying || to == kIsStopped ||
@@ -608,3 +626,38 @@ void AlsaPcmOutputStream::SharedData::set_volume(float v) {
   AutoLock l(lock_);
   volume_ = v;
 }
+
+size_t AlsaPcmOutputStream::SharedData::OnMoreData(AudioOutputStream* stream,
+                                                   void* dest,
+                                                   size_t max_size) {
+  AutoLock l(lock_);
+  if (source_callback_) {
+    return source_callback_->OnMoreData(stream, dest, max_size);
+  }
+
+  return 0;
+}
+
+void AlsaPcmOutputStream::SharedData::OnClose(AudioOutputStream* stream) {
+  AutoLock l(lock_);
+  if (source_callback_) {
+    source_callback_->OnClose(stream);
+  }
+}
+
+void AlsaPcmOutputStream::SharedData::OnError(AudioOutputStream* stream,
+                                              int code) {
+  AutoLock l(lock_);
+  if (source_callback_) {
+    source_callback_->OnError(stream, code);
+  }
+}
+
+// Changes the AudioSourceCallback to proxy calls to.  Pass in NULL to
+// release ownership of the currently registered callback.
+void AlsaPcmOutputStream::SharedData::set_source_callback(
+    AudioSourceCallback* callback) {
+  DCHECK_EQ(MessageLoop::current(), state_transition_loop_);
+  AutoLock l(lock_);
+  source_callback_ = callback;
+}
diff --git a/media/audio/linux/alsa_output.h b/media/audio/linux/alsa_output.h
index a36c1c2..7d5f381 100644
--- a/media/audio/linux/alsa_output.h
+++ b/media/audio/linux/alsa_output.h
@@ -40,6 +40,7 @@
 #include "testing/gtest/include/gtest/gtest_prod.h"
 
 class AlsaWrapper;
+class AudioManagerLinux;
 
 class AlsaPcmOutputStream :
     public AudioOutputStream,
@@ -63,6 +64,7 @@ class AlsaPcmOutputStream :
                       int sample_rate,
                       int bits_per_sample,
                       AlsaWrapper* wrapper,
+                      AudioManagerLinux* manager,
                       MessageLoop* message_loop);
   virtual ~AlsaPcmOutputStream();
 
@@ -120,7 +122,7 @@ class AlsaPcmOutputStream :
 
   // Various tasks that complete actions started in the public API.
   void FinishOpen(snd_pcm_t* playback_handle, size_t packet_size);
-  void StartTask(AudioSourceCallback* callback);
+  void StartTask();
   void CloseTask();
 
   // Functions to get another packet from the data source and write it into the
@@ -130,13 +132,6 @@ class AlsaPcmOutputStream :
   void WriteTask();
   void ScheduleNextWrite(Packet* current_packet);
 
-  // Functions to safeguard state transitions and ensure that transitions are
-  // only allowed occuring on the thread that created the object.  All changes
-  // to the object state should go through these functions.
-  bool CanTransitionTo(InternalState to);
-  bool CanTransitionTo_Locked(InternalState to);
-  InternalState TransitionTo(InternalState to);
-
   // Utility functions for talking with the ALSA API.
   static snd_pcm_sframes_t FramesInPacket(const Packet& packet,
                                           int bytes_per_frame);
@@ -145,6 +140,9 @@ class AlsaPcmOutputStream :
   bool CloseDevice(snd_pcm_t* handle);
   snd_pcm_sframes_t GetAvailableFrames();
 
+  // Thread-asserting accessors for member variables.
+  AudioManagerLinux* manager();
+
   // Struct holding all mutable the data that must be shared by the
   // message_loop() and the thread that created the object.
   class SharedData {
@@ -162,13 +160,33 @@ class AlsaPcmOutputStream :
     float volume();
     void set_volume(float v);
 
+    // API for Proxying calls to the AudioSourceCallback provided during
+    // Start().  These APIs are threadsafe.
+    //
+    // TODO(ajwong): This is necessary because the ownership semantics for the
+    // |source_callback_| object are incorrect in AudioRenderHost. The callback
+    // is passed into the output stream, but ownership is not transfered which
+    // requires a synchronization on access of the |source_callback_| to avoid
+    // using a deleted callback.
+    size_t OnMoreData(AudioOutputStream* stream, void* dest, size_t max_size);
+    void OnClose(AudioOutputStream* stream);
+    void OnError(AudioOutputStream* stream, int code);
+
+    // Changes the AudioSourceCallback to proxy calls to.  Pass in NULL to
+    // release ownership of the currently registered callback.
+    void set_source_callback(AudioSourceCallback* callback);
+
    private:
     Lock lock_;
 
     InternalState state_;
     float volume_;  // Volume level from 0.0 to 1.0.
 
+    AudioSourceCallback* source_callback_;
+
     MessageLoop* const state_transition_loop_;
+
+    DISALLOW_COPY_AND_ASSIGN(SharedData);
   } shared_data_;
 
   // Configuration constants from the constructor.  Referenceable by all threads
@@ -188,12 +206,14 @@ class AlsaPcmOutputStream :
   // Wrapper class to invoke all the ALSA functions.
   AlsaWrapper* wrapper_;
 
+  // Audio manager that created us.  Used to report that we've been closed.
+  // This should only be used on the |client_thread_loop_|.  Access via
+  // the manager() function.
+  AudioManagerLinux* manager_;
+
   // Handle to the actual PCM playback device.
   snd_pcm_t* playback_handle_;
 
-  // Callback used to request more data from the data source.
-  AudioSourceCallback* source_callback_;
-
   scoped_ptr<Packet> packet_;
   int frames_per_packet_;
 
diff --git a/media/audio/linux/alsa_output_unittest.cc b/media/audio/linux/alsa_output_unittest.cc
index 917c79f..8e262b1 100644
--- a/media/audio/linux/alsa_output_unittest.cc
+++ b/media/audio/linux/alsa_output_unittest.cc
@@ -5,6 +5,7 @@
 #include "base/logging.h"
 #include "media/audio/linux/alsa_output.h"
 #include "media/audio/linux/alsa_wrapper.h"
+#include "media/audio/linux/audio_manager_linux.h"
 #include "testing/gmock/include/gmock/gmock.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
@@ -45,6 +46,19 @@ class MockAudioSourceCallback : public AudioOutputStream::AudioSourceCallback {
   MOCK_METHOD2(OnError, void(AudioOutputStream* stream, int code));
 };
 
+class MockAudioManagerLinux : public AudioManagerLinux {
+ public:
+  MOCK_METHOD0(Init, void());
+  MOCK_METHOD0(HasAudioDevices, bool());
+  MOCK_METHOD4(MakeAudioStream, AudioOutputStream*(Format format, int channels,
+                                                   int sample_rate,
+                                                   char bits_per_sample));
+  MOCK_METHOD0(MuteAll, void());
+  MOCK_METHOD0(UnMuteAll, void());
+
+  MOCK_METHOD1(ReleaseStream, void(AlsaPcmOutputStream* stream));
+};
+
 class AlsaPcmOutputStreamTest : public testing::Test {
  protected:
   AlsaPcmOutputStreamTest()
@@ -55,6 +69,7 @@ class AlsaPcmOutputStreamTest : public testing::Test {
                                            kTestSampleRate,
                                            kTestBitsPerSample,
                                            &mock_alsa_wrapper_,
+                                           &mock_manager_,
                                            &message_loop_);
 
     packet_.size = kTestPacketSize;
@@ -77,6 +92,7 @@ class AlsaPcmOutputStreamTest : public testing::Test {
   static snd_pcm_t* const kFakeHandle;
 
   StrictMock<MockAlsaWrapper> mock_alsa_wrapper_;
+  StrictMock<MockAudioManagerLinux> mock_manager_;
   MessageLoop message_loop_;
   scoped_refptr<AlsaPcmOutputStream> test_stream_;
   AlsaPcmOutputStream::Packet packet_;
@@ -115,6 +131,7 @@ TEST_F(AlsaPcmOutputStreamTest, ConstructedState) {
                                          kTestSampleRate,
                                          kTestBitsPerSample,
                                          &mock_alsa_wrapper_,
+                                         &mock_manager_,
                                          &message_loop_);
   EXPECT_EQ(AlsaPcmOutputStream::kCreated,
             test_stream_->shared_data_.state());
@@ -126,6 +143,7 @@ TEST_F(AlsaPcmOutputStreamTest, ConstructedState) {
                                          kTestSampleRate,
                                          kTestBitsPerSample,
                                          &mock_alsa_wrapper_,
+                                         &mock_manager_,
                                          &message_loop_);
   EXPECT_EQ(AlsaPcmOutputStream::kInError,
             test_stream_->shared_data_.state());
@@ -137,6 +155,7 @@ TEST_F(AlsaPcmOutputStreamTest, ConstructedState) {
                                          kTestSampleRate,
                                          kTestBitsPerSample - 1,
                                          &mock_alsa_wrapper_,
+                                         &mock_manager_,
                                          &message_loop_);
   EXPECT_EQ(AlsaPcmOutputStream::kInError,
             test_stream_->shared_data_.state());
@@ -148,6 +167,7 @@ TEST_F(AlsaPcmOutputStreamTest, ConstructedState) {
                                          kTestSampleRate,
                                          kTestBitsPerSample,
                                          &mock_alsa_wrapper_,
+                                         &mock_manager_,
                                          &message_loop_);
   EXPECT_EQ(AlsaPcmOutputStream::kInError,
             test_stream_->shared_data_.state());
@@ -190,6 +210,7 @@ TEST_F(AlsaPcmOutputStreamTest, OpenClose) {
   // Now close it and test that everything was released.
   EXPECT_CALL(mock_alsa_wrapper_, PcmClose(kFakeHandle))
       .WillOnce(Return(0));
+  EXPECT_CALL(mock_manager_, ReleaseStream(test_stream_.get()));
   test_stream_->Close();
   message_loop_.RunAllPending();
 
@@ -266,6 +287,8 @@ TEST_F(AlsaPcmOutputStreamTest, StartStop) {
   test_stream_->Start(&mock_callback);
   message_loop_.RunAllPending();
 
+  EXPECT_CALL(mock_manager_, ReleaseStream(test_stream_.get()));
+  EXPECT_CALL(mock_callback, OnClose(test_stream_.get()));
   EXPECT_CALL(mock_alsa_wrapper_, PcmClose(kFakeHandle))
       .WillOnce(Return(0));
   test_stream_->Close();
@@ -343,7 +366,7 @@ TEST_F(AlsaPcmOutputStreamTest, BufferPacket) {
                          packet_.capacity))
       .WillOnce(Return(10));
 
-  test_stream_->source_callback_ = &mock_callback;
+  test_stream_->shared_data_.set_source_callback(&mock_callback);
   test_stream_->BufferPacket(&packet_);
 
   EXPECT_EQ(0u, packet_.used);
diff --git a/media/audio/linux/audio_manager_linux.cc b/media/audio/linux/audio_manager_linux.cc
index 3d1d98c..0c4db6a 100644
--- a/media/audio/linux/audio_manager_linux.cc
+++ b/media/audio/linux/audio_manager_linux.cc
@@ -42,9 +42,10 @@ AudioOutputStream* AudioManagerLinux::MakeAudioStream(Format format,
   AlsaPcmOutputStream* stream =
       new AlsaPcmOutputStream(AlsaPcmOutputStream::kDefaultDevice,
                               format, channels, sample_rate, bits_per_sample,
-                              wrapper_.get(), audio_thread_.message_loop());
+                              wrapper_.get(), this,
+                              audio_thread_.message_loop());
 
-  // TODO(ajwong): Set up this to clear itself when the stream closes.
+  AutoLock l(lock_);
   active_streams_[stream] = scoped_refptr<AlsaPcmOutputStream>(stream);
   return stream;
 }
@@ -55,6 +56,7 @@ AudioManagerLinux::AudioManagerLinux()
 }
 
 AudioManagerLinux::~AudioManagerLinux() {
+  active_streams_.clear();
 }
 
 void AudioManagerLinux::Init() {
@@ -63,15 +65,18 @@ void AudioManagerLinux::Init() {
 }
 
 void AudioManagerLinux::MuteAll() {
-  // TODO(ajwong): Implement.
   NOTIMPLEMENTED();
 }
 
 void AudioManagerLinux::UnMuteAll() {
-  // TODO(ajwong): Implement.
   NOTIMPLEMENTED();
 }
 
+void AudioManagerLinux::ReleaseStream(AlsaPcmOutputStream* stream) {
+  AutoLock l(lock_);
+  active_streams_.erase(stream);
+}
+
 // TODO(ajwong): Collapse this with the windows version.
 void DestroyAudioManagerLinux(void* not_used) {
   delete g_audio_manager;
diff --git a/media/audio/linux/audio_manager_linux.h b/media/audio/linux/audio_manager_linux.h
index d33c1a4..c79d319 100644
--- a/media/audio/linux/audio_manager_linux.h
+++ b/media/audio/linux/audio_manager_linux.h
@@ -20,7 +20,7 @@ class AudioManagerLinux : public AudioManager {
   AudioManagerLinux();
 
   // Call before using a newly created AudioManagerLinux instance.
-  void Init();
+  virtual void Init();
 
   // Implementation of AudioManager.
   virtual bool HasAudioDevices();
@@ -30,16 +30,20 @@ class AudioManagerLinux : public AudioManager {
   virtual void MuteAll();
   virtual void UnMuteAll();
 
- private:
-  // Friend function for invoking the private destructor at exit.
+  virtual void ReleaseStream(AlsaPcmOutputStream* stream);
+
+ protected:
+  // Friend function for invoking the destructor at exit.
   friend void DestroyAudioManagerLinux(void*);
   virtual ~AudioManagerLinux();
 
+ private:
   // Thread used to interact with AudioOutputStreams created by this
   // audio manger.
   base::Thread audio_thread_;
   scoped_ptr<AlsaWrapper> wrapper_;
 
+  Lock lock_;
   std::map<AlsaPcmOutputStream*, scoped_refptr<AlsaPcmOutputStream> >
       active_streams_;