There is a racing between SyncSocket::Receive in audio_thread_ and...

There is a racing between SyncSocket::Receive in audio_thread_ and SyncSocket::Close in renderer thread.
This patch fixes it by using a waitable event to signal the audio thread that it should stop.

Test=content_unittests by running Valgrind
BUG=103711

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

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

content/renderer/media/audio_device.cc

@@ -25,7 +25,8 @@ AudioDevice::AudioDevice(size_t buffer_size,
       callback_(callback),
       audio_delay_milliseconds_(0),
       volume_(1.0),
-      stream_id_(0) {
+      stream_id_(0),
+      memory_length_(0) {
   filter_ = RenderThreadImpl::current()->audio_message_filter();
   audio_data_.reserve(channels);
   for (int i = 0; i < channels; ++i) {
@@ -55,7 +56,8 @@ void AudioDevice::Start() {
       base::Bind(&AudioDevice::InitializeOnIOThread, this, params));
 }
 
-bool AudioDevice::Stop() {
+void AudioDevice::Stop() {
+  DCHECK(MessageLoop::current() != ChildProcess::current()->io_message_loop());
   // 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
   // the calling thread forever.
@@ -70,18 +72,19 @@ bool AudioDevice::Stop() {
   // 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
   // function call.
-  if (completion.TimedWait(kMaxTimeOut)) {
-    if (audio_thread_.get()) {
-      socket_->Close();
-      audio_thread_->Join();
-      audio_thread_.reset(NULL);
-    }
-  } else {
+  if (!completion.TimedWait(kMaxTimeOut)) {
     LOG(ERROR) << "Failed to shut down audio output on IO thread";
-    return false;
   }
 
-  return true;
+  if (audio_thread_.get()) {
+    // Close the socket handler to terminate the main thread function in the
+    // audio thread.
+    {
+      base::SyncSocket socket(socket_handle_);
+    }
+    audio_thread_->Join();
+    audio_thread_.reset(NULL);
+  }
 }
 
 bool AudioDevice::SetVolume(double volume) {
@@ -167,6 +170,7 @@ void AudioDevice::OnLowLatencyCreated(
   DCHECK_GE(socket_handle, 0);
 #endif
   DCHECK(length);
+  DCHECK(!audio_thread_.get());
 
   // Takes care of the case when Stop() is called before OnLowLatencyCreated().
   if (!stream_id_) {
@@ -176,14 +180,12 @@ void AudioDevice::OnLowLatencyCreated(
     return;
   }
 
-  shared_memory_.reset(new base::SharedMemory(handle, false));
-  shared_memory_->Map(length);
+  shared_memory_handle_ = handle;
+  memory_length_ = length;
 
   DCHECK_GE(length, buffer_size_ * sizeof(int16) * channels_);
 
-  socket_.reset(new base::SyncSocket(socket_handle));
-  // Allow the client to pre-populate the buffer.
-  FireRenderCallback();
+  socket_handle_ = socket_handle;
 
   audio_thread_.reset(
       new base::DelegateSimpleThread(this, "renderer_audio_thread"));
@@ -206,21 +208,28 @@ void AudioDevice::Send(IPC::Message* message) {
 void AudioDevice::Run() {
   audio_thread_->SetThreadPriority(base::kThreadPriority_RealtimeAudio);
 
+  base::SharedMemory shared_memory(shared_memory_handle_, false);
+  shared_memory.Map(memory_length_);
+  // Allow the client to pre-populate the buffer.
+  FireRenderCallback(reinterpret_cast<int16*>(shared_memory.memory()));
+
+  base::SyncSocket socket(socket_handle_);
+
   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;
 
-  while ((sizeof(pending_data) == socket_->Receive(&pending_data,
-                                                   sizeof(pending_data))) &&
+  while ((sizeof(pending_data) == socket.Receive(&pending_data,
+                                                 sizeof(pending_data))) &&
          (pending_data >= 0)) {
     // Convert the number of pending bytes in the render buffer
     // into milliseconds.
     audio_delay_milliseconds_ = pending_data / bytes_per_ms;
-    FireRenderCallback();
+    FireRenderCallback(reinterpret_cast<int16*>(shared_memory.memory()));
   }
 }
 
-void AudioDevice::FireRenderCallback() {
+void AudioDevice::FireRenderCallback(int16* data) {
   TRACE_EVENT0("audio", "AudioDevice::FireRenderCallback");
 
   if (callback_) {
@@ -229,7 +238,7 @@ void AudioDevice::FireRenderCallback() {
 
     // Interleave, scale, and clip to int16.
     media::InterleaveFloatToInt16(audio_data_,
-                                  static_cast<int16*>(shared_memory_data()),
+                                  data,
                                   buffer_size_);
   }
 }

content/renderer/media/audio_device.h

@@ -88,8 +88,8 @@ class CONTENT_EXPORT AudioDevice
   // Starts audio playback.
   void Start();
 
-  // Stops audio playback. Returns |true| on success.
-  bool Stop();
+  // Stops audio playback.
+  void Stop();
 
   // Sets the playback volume, with range [0.0, 1.0] inclusive.
   // Returns |true| on success.
@@ -127,7 +127,7 @@ class CONTENT_EXPORT AudioDevice
   // Method called on the audio thread (+ one call on the IO thread) ----------
   // Calls the client's callback for rendering audio. There will also be one
   // initial call on the IO thread before the audio thread has been created.
-  void FireRenderCallback();
+  void FireRenderCallback(int16* data);
 
   // DelegateSimpleThread::Delegate implementation.
   virtual void Run() OVERRIDE;
@@ -154,11 +154,6 @@ class CONTENT_EXPORT AudioDevice
   // Callbacks for rendering audio occur on this thread.
   scoped_ptr<base::DelegateSimpleThread> audio_thread_;
 
-  // IPC message stuff.
-  base::SharedMemory* shared_memory() { return shared_memory_.get(); }
-  base::SyncSocket* socket() { return socket_.get(); }
-  void* shared_memory_data() { return shared_memory()->memory(); }
-
   // Cached audio message filter (lives on the main render thread).
   scoped_refptr<AudioMessageFilter> filter_;
 
@@ -167,8 +162,9 @@ class CONTENT_EXPORT AudioDevice
 
   // Data transfer between browser and render process uses a combination
   // of sync sockets and shared memory to provide lowest possible latency.
-  scoped_ptr<base::SharedMemory> shared_memory_;
-  scoped_ptr<base::SyncSocket> socket_;
+  base::SharedMemoryHandle shared_memory_handle_;
+  base::SyncSocket::Handle socket_handle_;
+  int memory_length_;
 
   DISALLOW_IMPLICIT_CONSTRUCTORS(AudioDevice);
 };

content/renderer/media/audio_input_device.cc

@@ -25,7 +25,8 @@ AudioInputDevice::AudioInputDevice(size_t buffer_size,
       volume_(1.0),
       stream_id_(0),
       session_id_(0),
-      pending_device_ready_(false) {
+      pending_device_ready_(false),
+      memory_length_(0) {
   filter_ = RenderThreadImpl::current()->audio_input_message_filter();
   audio_data_.reserve(channels);
 #if defined(OS_MACOSX)
@@ -71,7 +72,8 @@ void AudioInputDevice::SetDevice(int session_id) {
                  session_id));
 }
 
-bool AudioInputDevice::Stop() {
+void AudioInputDevice::Stop() {
+  DCHECK(MessageLoop::current() != ChildProcess::current()->io_message_loop());
   VLOG(1) << "Stop()";
   // 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
@@ -88,21 +90,20 @@ bool AudioInputDevice::Stop() {
   // 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
   // function call.
-  if (completion.TimedWait(kMaxTimeOut)) {
-    if (audio_thread_.get()) {
-      // Terminate the main thread function in the audio thread.
-      socket_->Close();
-      // Wait for the audio thread to exit.
-      audio_thread_->Join();
-      // Ensures that we can call Stop() multiple times.
-      audio_thread_.reset(NULL);
-    }
-  } else {
+  if (!completion.TimedWait(kMaxTimeOut)) {
     LOG(ERROR) << "Failed to shut down audio input on IO thread";
-    return false;
   }
 
-  return true;
+  if (audio_thread_.get()) {
+    // Terminate the main thread function in the audio thread.
+    {
+      base::SyncSocket socket(socket_handle_);
+    }
+    // Wait for the audio thread to exit.
+    audio_thread_->Join();
+    // Ensures that we can call Stop() multiple times.
+    audio_thread_.reset(NULL);
+  }
 }
 
 bool AudioInputDevice::SetVolume(double volume) {
@@ -184,6 +185,7 @@ void AudioInputDevice::OnLowLatencyCreated(
   DCHECK_GE(socket_handle, 0);
 #endif
   DCHECK(length);
+  DCHECK(!audio_thread_.get());
 
   VLOG(1) << "OnLowLatencyCreated (stream_id=" << stream_id_ << ")";
   // Takes care of the case when Stop() is called before OnLowLatencyCreated().
@@ -194,10 +196,10 @@ void AudioInputDevice::OnLowLatencyCreated(
     return;
   }
 
-  shared_memory_.reset(new base::SharedMemory(handle, false));
-  shared_memory_->Map(length);
+  shared_memory_handle_ = handle;
+  memory_length_ = length;
 
-  socket_.reset(new base::SyncSocket(socket_handle));
+  socket_handle_ = socket_handle;
 
   audio_thread_.reset(
       new base::DelegateSimpleThread(this, "RendererAudioInputThread"));
@@ -225,7 +227,9 @@ void AudioInputDevice::OnStateChanged(AudioStreamState state) {
       // Joining the audio thread will be quite soon, since the stream has
       // been closed before.
       if (audio_thread_.get()) {
-        socket_->Close();
+        {
+          base::SyncSocket socket(socket_handle_);
+        }
         audio_thread_->Join();
         audio_thread_.reset(NULL);
       }
@@ -281,15 +285,20 @@ void AudioInputDevice::Send(IPC::Message* message) {
 void AudioInputDevice::Run() {
   audio_thread_->SetThreadPriority(base::kThreadPriority_RealtimeAudio);
 
+  base::SharedMemory shared_memory(shared_memory_handle_, false);
+  shared_memory.Map(memory_length_);
+
+  base::SyncSocket socket(socket_handle_);
+
   int pending_data;
   const int samples_per_ms =
       static_cast<int>(audio_parameters_.sample_rate) / 1000;
   const int bytes_per_ms = audio_parameters_.channels *
       (audio_parameters_.bits_per_sample / 8) * samples_per_ms;
 
-  while (sizeof(pending_data) == socket_->Receive(&pending_data,
-                                                  sizeof(pending_data)) &&
-                                                  pending_data >= 0) {
+  while ((sizeof(pending_data) == socket.Receive(&pending_data,
+                                                sizeof(pending_data))) &&
+         (pending_data >= 0)) {
     // TODO(henrika): investigate the provided |pending_data| value
     // and ensure that it is actually an accurate delay estimation.
 
@@ -297,18 +306,16 @@ void AudioInputDevice::Run() {
     // into milliseconds.
     audio_delay_milliseconds_ = pending_data / bytes_per_ms;
 
-    FireCaptureCallback();
+    FireCaptureCallback(reinterpret_cast<int16*>(shared_memory.memory()));
   }
 }
 
-void AudioInputDevice::FireCaptureCallback() {
+void AudioInputDevice::FireCaptureCallback(int16* input_audio) {
   if (!callback_)
       return;
 
   const size_t number_of_frames = audio_parameters_.samples_per_packet;
 
-  // Read 16-bit samples from shared memory (browser writes to it).
-  int16* input_audio = static_cast<int16*>(shared_memory_data());
   const int bytes_per_sample = sizeof(input_audio[0]);
 
   // Deinterleave each channel and convert to 32-bit floating-point

content/renderer/media/audio_input_device.h

@@ -131,9 +131,8 @@ class CONTENT_EXPORT AudioInputDevice
   void Start();
 
   // Stops audio capturing. This method is synchronous/blocking.
-  // Returns |true| on success.
   // TODO(henrika): add support for notification when recording has stopped.
-  bool Stop();
+  void Stop();
 
   // Sets the capture volume scaling, with range [0.0, 1.0] inclusive.
   // Returns |true| on success.
@@ -170,7 +169,7 @@ class CONTENT_EXPORT AudioInputDevice
 
   // Method called on the audio thread ----------------------------------------
   // Calls the client's callback for capturing audio.
-  void FireCaptureCallback();
+  void FireCaptureCallback(int16* input_audio);
 
   // DelegateSimpleThread::Delegate implementation.
   virtual void Run() OVERRIDE;
@@ -195,11 +194,6 @@ class CONTENT_EXPORT AudioInputDevice
   // Callbacks for capturing audio occur on this thread.
   scoped_ptr<base::DelegateSimpleThread> audio_thread_;
 
-  // IPC message stuff.
-  base::SharedMemory* shared_memory() { return shared_memory_.get(); }
-  base::SyncSocket* socket() { return socket_.get(); }
-  void* shared_memory_data() { return shared_memory()->memory(); }
-
   // Cached audio input message filter (lives on the main render thread).
   scoped_refptr<AudioInputMessageFilter> filter_;
 
@@ -214,8 +208,9 @@ class CONTENT_EXPORT AudioInputDevice
   // callback. Only modified on the IO thread.
   bool pending_device_ready_;
 
-  scoped_ptr<base::SharedMemory> shared_memory_;
-  scoped_ptr<base::SyncSocket> socket_;
+  base::SharedMemoryHandle shared_memory_handle_;
+  base::SyncSocket::Handle socket_handle_;
+  int memory_length_;
 
   DISALLOW_IMPLICIT_CONSTRUCTORS(AudioInputDevice);
 };

content/renderer/media/webrtc_audio_device_impl.cc

@@ -597,8 +597,9 @@ int32_t WebRtcAudioDeviceImpl::StopPlayout() {
     // webrtc::VoiceEngine assumes that it is OK to call Stop() just in case.
     return 0;
   }
-  playing_ = !audio_output_device_->Stop();
-  return (!playing_ ? 0 : -1);
+  audio_output_device_->Stop();
+  playing_ = false;
+  return 0;
 }
 
 bool WebRtcAudioDeviceImpl::Playing() const {
@@ -646,8 +647,9 @@ int32_t WebRtcAudioDeviceImpl::StopRecording() {
     // webrtc::VoiceEngine assumes that it is OK to call Stop() just in case.
     return 0;
   }
-  recording_ = !audio_input_device_->Stop();
-  return (!recording_ ? 0 : -1);
+  audio_input_device_->Stop();
+  recording_ = false;
+  return 0;
 }
 
 bool WebRtcAudioDeviceImpl::Recording() const {

content/renderer/renderer_webaudiodevice_impl.cc

@@ -27,8 +27,8 @@ void RendererWebAudioDeviceImpl::start() {
 
 void RendererWebAudioDeviceImpl::stop() {
   if (is_running_) {
-    if (audio_device_->Stop())
-      is_running_ = false;
+    audio_device_->Stop();
+    is_running_ = false;
   }
 }