Switch AudioDevice classes from SyncSocket to CancelableSyncSocket.

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/bind.h"
 #include "base/message_loop.h"
+#include "base/threading/thread_restrictions.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_impl.h"
 #include "media/audio/audio_manager_base.h"
 #include "media/audio/audio_util.h"
 
 AudioInputDevice::AudioInputDevice(size_t buffer_size,
                                    int channels,
                                    double sample_rate,
                                    CaptureCallback* callback,
                                    CaptureEventHandler* event_handler)
     : callback_(callback),
       event_handler_(event_handler),
       audio_delay_milliseconds_(0),
       volume_(1.0),
       stream_id_(0),
       session_id_(0),
       pending_device_ready_(false),
       shared_memory_handle_(base::SharedMemory::NULLHandle()),
-      socket_handle_(base::SyncSocket::kInvalidHandle),
       memory_length_(0) {
   filter_ = RenderThreadImpl::current()->audio_input_message_filter();
   audio_data_.reserve(channels);
 #if defined(OS_MACOSX)
   VLOG(1) << "Using AUDIO_PCM_LOW_LATENCY as input mode on Mac OS X.";
   audio_parameters_.format = AudioParameters::AUDIO_PCM_LOW_LATENCY;
 #elif defined(OS_WIN)
   VLOG(1) << "Using AUDIO_PCM_LOW_LATENCY as input mode on Windows.";
   audio_parameters_.format = AudioParameters::AUDIO_PCM_LOW_LATENCY;
 #else
@@ skipping 29 matching lines @@
   VLOG(1) << "SetDevice (session_id=" << session_id << ")";
   ChildProcess::current()->io_message_loop()->PostTask(
       FROM_HERE,
       base::Bind(&AudioInputDevice::SetSessionIdOnIOThread, this,
                  session_id));
 }
 
 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
-  // the calling thread forever.
-  const base::TimeDelta kMaxTimeOut = base::TimeDelta::FromMilliseconds(1000);
 
   base::WaitableEvent completion(false, false);
-
   ChildProcess::current()->io_message_loop()->PostTask(
       FROM_HERE,
       base::Bind(&AudioInputDevice::ShutDownOnIOThread, this,
                  &completion));
 
   // 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)) {
-    LOG(ERROR) << "Failed to shut down audio input on IO thread";
-  }
-
-  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);
-  }
+  completion.Wait();
 }
 
 bool AudioInputDevice::SetVolume(double volume) {
   NOTIMPLEMENTED();
   return false;
 }
 
 bool AudioInputDevice::GetVolume(double* volume) {
   NOTIMPLEMENTED();
   return false;
@@ skipping 35 matching lines @@
   DCHECK(MessageLoop::current() == ChildProcess::current()->io_message_loop());
   // Make sure we don't call shutdown more than once.
   if (!stream_id_) {
     completion->Signal();
     return;
   }
 
   filter_->RemoveDelegate(stream_id_);
   Send(new AudioInputHostMsg_CloseStream(stream_id_));
 
+  ShutDownAudioThread();
+
   stream_id_ = 0;
   session_id_ = 0;
   pending_device_ready_ = false;
 
   completion->Signal();
 }
 
 void AudioInputDevice::SetVolumeOnIOThread(double volume) {
   DCHECK(MessageLoop::current() == ChildProcess::current()->io_message_loop());
   if (stream_id_)
@@ skipping 19 matching lines @@
   // Takes care of the case when Stop() is called before OnLowLatencyCreated().
   if (!stream_id_) {
     base::SharedMemory::CloseHandle(handle);
     // Close the socket handler.
     base::SyncSocket socket(socket_handle);
     return;
   }
 
   shared_memory_handle_ = handle;
   memory_length_ = length;
-
-  socket_handle_ = socket_handle;
+  audio_socket_.reset(new base::CancelableSyncSocket(socket_handle));
 
   audio_thread_.reset(
       new base::DelegateSimpleThread(this, "RendererAudioInputThread"));
   audio_thread_->Start();
 
   MessageLoop::current()->PostTask(
       FROM_HERE,
       base::Bind(&AudioInputDevice::StartOnIOThread, this));
 }
 
 void AudioInputDevice::OnVolume(double volume) {
   NOTIMPLEMENTED();
 }
 
 void AudioInputDevice::OnStateChanged(AudioStreamState state) {
   DCHECK(MessageLoop::current() == ChildProcess::current()->io_message_loop());
   switch (state) {
     case kAudioStreamPaused:
+      // TODO(xians): Should we just call ShutDownOnIOThread here instead?
+
       // Do nothing if the stream has been closed.
       if (!stream_id_)
         return;
 
       filter_->RemoveDelegate(stream_id_);
 
-      // Joining the audio thread will be quite soon, since the stream has
-      // been closed before.
-      if (audio_thread_.get()) {
-        {
-          base::SyncSocket socket(socket_handle_);
-        }
-        audio_thread_->Join();
-        audio_thread_.reset(NULL);
-      }
+      ShutDownAudioThread();
 
       if (event_handler_)
         event_handler_->OnDeviceStopped();
 
       stream_id_ = 0;
       pending_device_ready_ = false;
       break;
     case kAudioStreamPlaying:
       NOTIMPLEMENTED();
       break;
@@ skipping 13 matching lines @@
   // Takes care of the case when Stop() is called before OnDeviceReady().
   if (!pending_device_ready_)
     return;
 
   // If AudioInputDeviceManager returns an empty string, it means no device
   // is ready for start.
   if (device_id.empty()) {
     filter_->RemoveDelegate(stream_id_);
     stream_id_ = 0;
   } else {
-    Send(new AudioInputHostMsg_CreateStream(
-        stream_id_, audio_parameters_, true, device_id));
+    Send(new AudioInputHostMsg_CreateStream(stream_id_, audio_parameters_,
+                                            true, device_id));
   }
 
   pending_device_ready_ = false;
   // Notify the client that the device has been started.
   if (event_handler_)
     event_handler_->OnDeviceStarted(device_id);
 }
 
 void AudioInputDevice::Send(IPC::Message* message) {
   filter_->Send(message);
 }
 
 // Our audio thread runs here. We receive captured audio samples on
 // this thread.
 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_);
+  base::CancelableSyncSocket* socket = audio_socket_.get();

[scherkus (not reviewing), 2012/01/30 19:10:57]

nit: do we need this line of code?

[tommi (sloooow) - chröme, 2012/01/30 21:53:14]

yup (different than before)

 
-  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 (true) {
+    uint32 pending_data = 0;
+    size_t received = socket->Receive(&pending_data, sizeof(pending_data));
+    if (received != sizeof(pending_data)) {
+      DCHECK(received == 0U);
+      break;
+    }
     // TODO(henrika): investigate the provided |pending_data| value
     // and ensure that it is actually an accurate delay estimation.
 
     // Convert the number of pending bytes in the capture buffer
     // into milliseconds.
     audio_delay_milliseconds_ = pending_data / bytes_per_ms;
 
     FireCaptureCallback(reinterpret_cast<int16*>(shared_memory.memory()));
   }
 }
 
 void AudioInputDevice::FireCaptureCallback(int16* input_audio) {
   if (!callback_)
-      return;
+    return;
 
   const size_t number_of_frames = audio_parameters_.samples_per_packet;
 
   const int bytes_per_sample = sizeof(input_audio[0]);
 
   // Deinterleave each channel and convert to 32-bit floating-point
   // with nominal range -1.0 -> +1.0.
   for (int channel_index = 0; channel_index < audio_parameters_.channels;
        ++channel_index) {
     media::DeinterleaveAudioChannel(input_audio,
                                     audio_data_[channel_index],
                                     audio_parameters_.channels,
                                     channel_index,
                                     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::ShutDownAudioThread() {
+  DCHECK_EQ(MessageLoop::current(), ChildProcess::current()->io_message_loop());
+
+  // TODO(tommi): This is a copy/paste of the same method in AudioDevice.
+  // We could extract the worker thread+socket bits out to a separate
+  // class to avoid having to fix the same issues twice in these classes.
+
+  if (audio_thread_.get()) {
+    // Close the socket to terminate the main thread function in the
+    // audio thread.
+    audio_socket_->Shutdown();  // Stops blocking Receive calls.
+    // TODO(tommi): We must not do this from the IO thread.  Fix.
+    base::ThreadRestrictions::ScopedAllowIO allow_wait;
+    audio_thread_->Join();
+    audio_thread_.reset(NULL);
+    audio_socket_.reset();
+  }
+}