Refactor AudioInputDevice to remove race conditions and allow more flexible calling sequences.

content/renderer/media/audio_input_device.cc

 // Copyright (c) 2011 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 "content/renderer/media/audio_input_device.h"
 
 #include "base/message_loop.h"
+#include "base/time.h"
 #include "content/common/child_process.h"
 #include "content/common/media/audio_messages.h"
 #include "content/common/view_messages.h"
 #include "content/renderer/render_thread.h"
 #include "media/audio/audio_util.h"
 
+// Max waiting time for Stop to complete. If this time limit is passed,
+// we will stop waiting and return false. It ensures that Stop can't block

[scherkus (not reviewing), 2011/08/23 16:54:19]

Stop -> Stop()

[henrika (OOO until Aug 14), 2011/08/24 15:03:55]

Done.

+// the calling thread forever.
+static const base::TimeDelta kMaxTimeOutMs =

[scherkus (not reviewing), 2011/08/23 16:54:19]

nit: drop Ms suffix since TimeDelats are unitless

[henrika (OOO until Aug 14), 2011/08/24 15:03:55]

Thanks. Done.

+    base::TimeDelta::FromMilliseconds(1000);
+
 AudioInputDevice::AudioInputDevice(size_t buffer_size,
                                    int channels,
                                    double sample_rate,
                                    CaptureCallback* callback)
     : buffer_size_(buffer_size),
       channels_(channels),
       bits_per_sample_(16),
       sample_rate_(sample_rate),
       callback_(callback),
       audio_delay_milliseconds_(0),
       volume_(1.0),
       stream_id_(0) {
   filter_ = RenderThread::current()->audio_input_message_filter();
   audio_data_.reserve(channels);
   for (int i = 0; i < channels; ++i) {
     float* channel_data = new float[buffer_size];
     audio_data_.push_back(channel_data);
   }
 }
 
 AudioInputDevice::~AudioInputDevice() {
-  // Make sure we have been shut down.
-  DCHECK_EQ(0, stream_id_);
-  Stop();
+  // TODO(henrika): The current design requires that the user calls
+  // Stop before deleting this class.
+  CHECK_EQ(0, stream_id_);
   for (int i = 0; i < channels_; ++i)
     delete [] audio_data_[i];
 }
 
-bool AudioInputDevice::Start() {
-  // Make sure we don't call Start() more than once.
-  DCHECK_EQ(0, stream_id_);
-  if (stream_id_)
-    return false;
-
+void AudioInputDevice::Start() {
+  VLOG(1) << "Start()";
   AudioParameters params;
   // TODO(henrika): add support for low-latency mode?
   params.format = AudioParameters::AUDIO_PCM_LINEAR;
   params.channels = channels_;
   params.sample_rate = static_cast<int>(sample_rate_);
   params.bits_per_sample = bits_per_sample_;
   params.samples_per_packet = buffer_size_;
 
   ChildProcess::current()->io_message_loop()->PostTask(
-    FROM_HERE,
-    NewRunnableMethod(this, &AudioInputDevice::InitializeOnIOThread, params));
-
-  return true;
+      FROM_HERE,
+      NewRunnableMethod(this, &AudioInputDevice::InitializeOnIOThread, params));
 }
 
 bool AudioInputDevice::Stop() {
-  if (!stream_id_)
-    return false;
+  VLOG(1) << "Stop()";
+  base::WaitableEvent completion(false, false);
 
   ChildProcess::current()->io_message_loop()->PostTask(
       FROM_HERE,
-      NewRunnableMethod(this, &AudioInputDevice::ShutDownOnIOThread));
+      NewRunnableMethod(this, &AudioInputDevice::ShutDownOnIOThread,
+                        &completion));
 
-  if (audio_thread_.get()) {
-    socket_->Close();
-    audio_thread_->Join();
+  // We wait here for the IO task to be completed to remove race conflicts
+  // with OnLowLatencyCreated and to ensure that Stop acts as a synchronous

[scherkus (not reviewing), 2011/08/23 16:54:19]

don't forget () suffix on function call names in comments

[henrika (OOO until Aug 14), 2011/08/24 15:03:55]

Done.

+  // function call.
+  if (completion.TimedWait(kMaxTimeOutMs)) {

[scherkus (not reviewing), 2011/08/23 16:54:19]

out of curiosity who is usually the calling thread?

do we know how fast this executes under normal circumstances?

[henrika (OOO until Aug 14), 2011/08/24 15:03:55]

The calling client will be calling from the main render thread . Would work with
any worker thread actually. The timeout is probably overkill since the main
thread loop should be done when Join is called, hence the delay should be very
low. I added the timeout just in case.

+    if (audio_thread_.get()) {
+      // Terminate the main thread function in the audio thread.
+      socket_->Close();
+      // Wait for the audio thread to exit.

[scherkus (not reviewing), 2011/08/23 16:54:19]

nit: most of these comments are useless as you have a block-level comment above
describing what's going on

+      audio_thread_->Join();
+      // Ensures that we can call Stop multiple times.

[scherkus (not reviewing), 2011/08/23 16:54:19]

who's calling Stop() multiple times?

[henrika (OOO until Aug 14), 2011/08/24 15:03:55]

Well, we just want to ensure that the design can deal with it if a client (that
we can't control) calls Stop, Stop. This is one way of ensuring that the second
call "does nothing" and then returns true. Without the audio_thread_.reset(NULL)
we can't call Start()....Stop(),Stop() without crashing since we will try to
join a thread multiple times.

+      audio_thread_.reset(NULL);
+    }
+  } else {
+    DLOG(ERROR) << "Failed to shut down audio input on IO thread";

[scherkus (not reviewing), 2011/08/23 16:54:19]

how often do we expect to see this in practice?

might want to bump it up to LOG() if it's a serious enough issue

[henrika (OOO until Aug 14), 2011/08/24 15:03:55]

Very good question. Classic comment: "should never happen". Will add to LOG().

+    return false;
   }
 
   return true;
 }
 
 bool AudioInputDevice::SetVolume(double volume) {
   NOTIMPLEMENTED();
   return false;
 }
 
 bool AudioInputDevice::GetVolume(double* volume) {
   NOTIMPLEMENTED();
   return false;
 }
 
 void AudioInputDevice::InitializeOnIOThread(const AudioParameters& params) {
+  DCHECK(MessageLoop::current() == ChildProcess::current()->io_message_loop());
+  // Make sure we don't call Start() more than once.
+  DCHECK_EQ(0, stream_id_);
+  if (stream_id_)
+    return;
+
   stream_id_ = filter_->AddDelegate(this);
   Send(new AudioInputHostMsg_CreateStream(stream_id_, params, true));
 }
 
 void AudioInputDevice::StartOnIOThread() {
+  DCHECK(MessageLoop::current() == ChildProcess::current()->io_message_loop());
   if (stream_id_)
     Send(new AudioInputHostMsg_RecordStream(stream_id_));
 }
 
-void AudioInputDevice::ShutDownOnIOThread() {
+void AudioInputDevice::ShutDownOnIOThread(base::WaitableEvent* completion) {
+  DCHECK(MessageLoop::current() == ChildProcess::current()->io_message_loop());
   // Make sure we don't call shutdown more than once.
-  if (!stream_id_)
+  if (!stream_id_) {
+    completion->Signal();
     return;
+  }
 
   filter_->RemoveDelegate(stream_id_);
   Send(new AudioInputHostMsg_CloseStream(stream_id_));
   stream_id_ = 0;
+
+  completion->Signal();
 }
 
 void AudioInputDevice::SetVolumeOnIOThread(double volume) {
+  DCHECK(MessageLoop::current() == ChildProcess::current()->io_message_loop());
   if (stream_id_)
     Send(new AudioInputHostMsg_SetVolume(stream_id_, volume));
 }
 
 void AudioInputDevice::OnLowLatencyCreated(
     base::SharedMemoryHandle handle,
     base::SyncSocket::Handle socket_handle,
     uint32 length) {
   DCHECK(MessageLoop::current() == ChildProcess::current()->io_message_loop());
 #if defined(OS_WIN)
   DCHECK(handle);
   DCHECK(socket_handle);
 #else
   DCHECK_GE(handle.fd, 0);
   DCHECK_GE(socket_handle, 0);
 #endif
   DCHECK(length);
 
+  VLOG(1) << __FUNCTION__ << "[stream_id=" << stream_id_ << "]";

[scherkus (not reviewing), 2011/08/23 16:54:19]

use OnLowLatencyCreated ?

[henrika (OOO until Aug 14), 2011/08/24 15:03:55]

Done.

+  // Takes care of the case when Stop is called before OnLowLatencyCreated.
+  if (!stream_id_) {
+    base::SharedMemory::CloseHandle(handle);
+    base::SyncSocket socket(socket_handle);
+    return;
+  }
+
   shared_memory_.reset(new base::SharedMemory(handle, false));
   shared_memory_->Map(length);
 
   socket_.reset(new base::SyncSocket(socket_handle));
 
   audio_thread_.reset(
-      new base::DelegateSimpleThread(this, "renderer_audio_input_thread"));
+      new base::DelegateSimpleThread(this, "RendererAudioInputThread"));
   audio_thread_->Start();
 
   MessageLoop::current()->PostTask(
       FROM_HERE,
       NewRunnableMethod(this, &AudioInputDevice::StartOnIOThread));
 }
 
 void AudioInputDevice::OnVolume(double volume) {
   NOTIMPLEMENTED();
 }
@@ skipping 45 matching lines @@
                                     bytes_per_sample,
                                     number_of_frames);
   }
 
   // Deliver captured data to the client in floating point format
   // and update the audio-delay measurement.
   callback_->Capture(audio_data_,
                      number_of_frames,
                      audio_delay_milliseconds_);
 }