refactor AudioInputDevice to remove race condition.

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 "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"
 
 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),
+      capture_thread_("AudioInputDevice"),

[henrika_dont_use, 2011/07/30 15:50:59]

Do we use camel-back style for thread names?

[wjia(left Chromium), 2011/08/01 22:40:13]

Yes.

+      delegate_event_(false, false),
+      stop_event_(false, false),
       stream_id_(0) {
   filter_ = RenderThread::current()->audio_input_message_filter();
+  capture_thread_.Start();
+  message_loop_proxy_ = capture_thread_.message_loop_proxy();

[henrika_dont_use, 2011/07/30 15:50:59]

Sorry, again, why not message_loop()?
Also, why is is called such a generic name? It might be more clear to give it a
name so we know that tasks on this thread runs on the main/capture thread. Just
and  idea; perhaps your style is the preferred in Chrome.

[wjia(left Chromium), 2011/08/01 22:40:13]

In audio_input_device.h, the difference between MessageLoop and MessageLoopProxy
has been explained.
message_loop_proxy_ is renamed to capture_message_loop_proxy_.

   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();
+  capture_thread_.Stop();

[henrika_dont_use, 2011/07/30 15:50:59]

You did not like DCHECK_EQ(0, stream_id_)?

[xians, 2011/08/01 11:41:23]

It is for thread safety, stream_id_ is only accessed on the capture thread.

On 2011/07/30 15:50:59, henrika wrote:

[wjia(left Chromium), 2011/08/01 22:40:13]

Shijing has the answer.

   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;
-
-  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));
-
+  message_loop_proxy_->PostTask(FROM_HERE,

[henrika_dont_use, 2011/07/30 15:50:59]

Why bool if false is not an option?

[wjia(left Chromium), 2011/08/01 22:40:13]

I didn't change ADM. ADM should be refactored as well.

+      NewRunnableMethod(this, &AudioInputDevice::StartOnCaptureThread));
   return true;
 }
 
 bool AudioInputDevice::Stop() {
-  if (!stream_id_)
-    return false;
-
-  ChildProcess::current()->io_message_loop()->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(this, &AudioInputDevice::ShutDownOnIOThread));
-
-  if (audio_thread_.get()) {
-    socket_->Close();
-    audio_thread_->Join();
-  }
-
+  message_loop_proxy_->PostTask(FROM_HERE,

[henrika_dont_use, 2011/07/30 15:50:59]

Why bool?

[wjia(left Chromium), 2011/08/01 22:40:13]

Same reason for Start().

+      NewRunnableMethod(this, &AudioInputDevice::ShutDownOnCaptureThread));
+  stop_event_.Wait();
   return true;
 }
 
 bool AudioInputDevice::SetVolume(double volume) {
   NOTIMPLEMENTED();
   return false;
 }
 
 bool AudioInputDevice::GetVolume(double* volume) {
   NOTIMPLEMENTED();
   return false;
 }
 
-void AudioInputDevice::InitializeOnIOThread(const AudioParameters& params) {
-  stream_id_ = filter_->AddDelegate(this);
+void AudioInputDevice::StartOnCaptureThread() {
+  // Make sure we don't call Start() more than once.

[henrika_dont_use, 2011/07/30 15:50:59]

Must we check stream_id on the capture thread? Could it not be checked one level
above this method?

[xians, 2011/08/01 11:41:23]

ditto, thread safety.

On 2011/07/30 15:50:59, henrika wrote:

+  if (stream_id_)
+    return;
+
+  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_proxy()->PostTask(FROM_HERE,

[henrika_dont_use, 2011/07/30 15:50:59]

why proxy (again) ;-) ? It was jam who initially suggested ->io_message_loop()->

[wjia(left Chromium), 2011/08/01 22:40:13]

MessageLoopProxy is preferred over MessageLoop.

+      NewRunnableMethod(this, &AudioInputDevice::AddDelegateOnIOThread));
+  delegate_event_.Wait();

[henrika_dont_use, 2011/07/30 15:50:59]

I have only done a quick check so far but it feels like the new Start design is
"not as clean" as the old version. The fact that this event is needed is one
indication. Was there actually a potential race in the old version? Or is the
new event a requirement now when the new worker thread has been added?

My point is, it looks OK but was the current change really required for Start? I
am not saying that we should change.

[xians, 2011/08/01 11:41:23]

My understanding is that this delegate_event.Wait() in the new Start() is used
to sync with the IO thread on waiting for stream_id_ to be updated.  Maybe Wei
have something to add here.
Yes, there was a potential race in the old version. Calling Start/Stop in
sequence is an example.

On 2011/07/30 15:50:59, henrika wrote:

[wjia(left Chromium), 2011/08/01 22:40:13]

Shijing has the answer. The delegate has to be added on IO thread (without
lock). Without delegate_event_, it's needed to have some message posting back
and forth between capture_thread_ and io thread, just like what's done in
VideoCaptureImpl. That will result in a much bigger change.

+
   Send(new AudioInputHostMsg_CreateStream(stream_id_, params, true));

[henrika_dont_use, 2011/07/30 15:50:59]

Add comment about that true<=>low-latency audio stream.

[wjia(left Chromium), 2011/08/01 22:40:13]

?? There was no comment for this.

 }
 
-void AudioInputDevice::StartOnIOThread() {
-  if (stream_id_)
-    Send(new AudioInputHostMsg_RecordStream(stream_id_));
-}
-
-void AudioInputDevice::ShutDownOnIOThread() {
-  // Make sure we don't call shutdown more than once.
-  if (!stream_id_)
-    return;
-
-  filter_->RemoveDelegate(stream_id_);
-  Send(new AudioInputHostMsg_CloseStream(stream_id_));
-  stream_id_ = 0;
-}
-
-void AudioInputDevice::SetVolumeOnIOThread(double volume) {
-  if (stream_id_)
-    Send(new AudioInputHostMsg_SetVolume(stream_id_, volume));
-}
-
-void AudioInputDevice::OnLowLatencyCreated(
+void AudioInputDevice::StartRecordingOnCaptureThread(
     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);
 
+  if (!stream_id_) {

[henrika_dont_use, 2011/07/30 15:50:59]

How could this happen? If so, can't we detect it earlier?

[wjia(left Chromium), 2011/08/01 22:40:13]

When Stop is called before OnLowLatencyCreated is called. This is rare, but
still a corner case.

+    base::SharedMemory::CloseHandle(handle);
+    base::SyncSocket socket(socket_handle);

[xians, 2011/08/01 11:41:23]

Sorry, the same question as Henrik has, + why do we need:
base::SharedMemory::CloseHandle(handle);
base::SyncSocket socket(socket_handle);

[wjia(left Chromium), 2011/08/01 22:40:13]

both handles should be closed.

+    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"));

[henrika_dont_use, 2011/07/30 15:50:59]

Note style for thread name and compare with AudioInputDevice thread.

[wjia(left Chromium), 2011/08/01 22:40:13]

I missed this one copied from previous code. Changed it to camelCase.

   audio_thread_->Start();
 
-  MessageLoop::current()->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(this, &AudioInputDevice::StartOnIOThread));
+  Send(new AudioInputHostMsg_RecordStream(stream_id_));
+}
+
+void AudioInputDevice::ShutDownOnCaptureThread() {
+  // Make sure we don't call shutdown more than once.
+  if (!stream_id_) {
+    stop_event_.Signal();
+    return;
+  }
+
+  ChildProcess::current()->io_message_loop_proxy()->PostTask(FROM_HERE,
+      NewRunnableMethod(this, &AudioInputDevice::RemoveDelegateOnIOThread,
+                        stream_id_));
+  Send(new AudioInputHostMsg_CloseStream(stream_id_));

[henrika_dont_use, 2011/07/30 15:50:59]

We wait here on Start. Is it not suitable to do so on Stop as well? If not,
perhaps a comment to explain why not.

Or, perhaps the comment should be in Start to explain why we must wait for the
stream_id. Hope my comment makes sense.

[wjia(left Chromium), 2011/08/01 22:40:13]

I moved Wait to a little bit later to allow IO thread and audio_thread_ execute
something in parallel. Just a matter of optimization to save some time.

+  stream_id_ = 0;
+
+  if (audio_thread_.get()) {

[xians, 2011/08/01 11:41:23]

shall we have a state here instead?
Think about the following scenario:
audio_thread_ was joined, then user call start/stop in sequence, then we may get
into this ShutDownOnCaptureThread() before OnLowLatencyCreated(), then we will
call join() twice, which may crash the browser.
Or OnLowLatencyCreated() never comes, and user calls Stop, it will also join the
audio_thread_.

[wjia(left Chromium), 2011/08/01 22:40:13]

Immediately after audio_thread_ is joined, it's set to NULL and will not be
joined again.

+    socket_->Close();
+    audio_thread_->Join();
+    audio_thread_.reset(NULL);

[henrika_dont_use, 2011/07/30 15:50:59]

Is NULL really needed here?

[wjia(left Chromium), 2011/08/01 22:40:13]

yes, in Shijing's exemplary use case, 2nd Shutdown can happen before 2nd
OnLowLatencyCreated is called. 

+  }
+
+  delegate_event_.Wait();

[henrika_dont_use, 2011/07/30 15:50:59]

OK, you wait here instead. Hmm. Not sure why.

[wjia(left Chromium), 2011/08/01 22:40:13]

Explained as above.

+  stop_event_.Signal();
+}
+
+void AudioInputDevice::SetVolumeOnCaptureThread(double volume) {
+  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());
+  message_loop_proxy_->PostTask(FROM_HERE,
+      NewRunnableMethod(this, &AudioInputDevice::StartRecordingOnCaptureThread,
+                        handle, socket_handle, length));
 }
 
 void AudioInputDevice::OnVolume(double volume) {
   NOTIMPLEMENTED();
 }
 
 void AudioInputDevice::Send(IPC::Message* message) {
-  filter_->Send(message);
+  ChildProcess::current()->io_message_loop_proxy()->PostTask(FROM_HERE,

[henrika_dont_use, 2011/07/30 15:50:59]

It feels more clear to name this SendOnIOThread() now when it is called
differently (not only on the IO as before)?

[wjia(left Chromium), 2011/08/01 22:40:13]

Done.

+      NewRunnableMethod(filter_.get(),
+                        &AudioInputMessageFilter::Send, message));
 }
 
 // Our audio thread runs here. We receive captured audio samples on
 // this thread.
 void AudioInputDevice::Run() {
   audio_thread_->SetThreadPriority(base::kThreadPriority_RealtimeAudio);
 
   int pending_data;
   const int samples_per_ms = static_cast<int>(sample_rate_) / 1000;
   const int bytes_per_ms = channels_ * (bits_per_sample_ / 8) * samples_per_ms;
@@ skipping 32 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_);
 }
+
+void AudioInputDevice::AddDelegateOnIOThread() {
+  stream_id_ = filter_->AddDelegate(this);

[henrika_dont_use, 2011/07/30 15:50:59]

Would be nice with a VLOG(1) here. Feels great to be able to see what stream IDs
we create.

[wjia(left Chromium), 2011/08/01 22:40:13]

Done.

+  delegate_event_.Signal();
+}
+
+void AudioInputDevice::RemoveDelegateOnIOThread(int32 stream_id) {
+  filter_->RemoveDelegate(stream_id);

[henrika_dont_use, 2011/07/30 15:50:59]

VLOG(1) please.

[wjia(left Chromium), 2011/08/01 22:40:13]

Done.

+  delegate_event_.Signal();
+}