Remove task_runner from audio_input_device, as the ipc object is threadsafe now

Bug: chromium:672469
Cq-Include-Trybots: luci.chromium.try:android_optional_gpu_tests_rel;luci.chromium.try:linux_optional_gpu_tests_rel;luci.chromium.try:mac_optional_gpu_tests_rel;luci.chromium.try:win_optional_gpu_tests_rel
Change-Id: I1e24e99df4a6d9dc0480aeb460115bafdceeafd5
Reviewed-on: https://chromium-review.googlesource.com/1018475
Commit-Queue: Jonas Olsson <jonasolsson@chromium.org>
Reviewed-by: John Abd-El-Malek <jam@chromium.org>
Reviewed-by: Max Morin <maxmorin@chromium.org>
Reviewed-by: Olga Sharonova <olka@chromium.org>
Cr-Commit-Position: refs/heads/master@{#553499}

base/threading/thread_restrictions.h

@@ -74,6 +74,7 @@ namespace leveldb {
 class LevelDBMojoProxy;
 }
 namespace media {
+class AudioInputDevice;
 class BlockingUrlProtocol;
 }
 namespace midi {
@@ -217,6 +218,7 @@ class BASE_EXPORT ScopedAllowBlocking {
   friend class content::BrowserProcessSubThread;
   friend class cronet::CronetPrefsManager;
   friend class cronet::CronetURLRequestContext;
+  friend class media::AudioInputDevice;
   friend class mojo::CoreLibraryInitializer;
   friend class resource_coordinator::TabManagerDelegate;  // crbug.com/778703
   friend class ui::MaterialDesignController;

content/renderer/media/audio_device_factory.cc

@@ -165,8 +165,7 @@ AudioDeviceFactory::NewAudioCapturerSource(int render_frame_id) {
   }
 
   return base::MakeRefCounted<media::AudioInputDevice>(
-      AudioInputIPCFactory::get()->CreateAudioInputIPC(render_frame_id),
-      AudioInputIPCFactory::get()->io_task_runner());
+      AudioInputIPCFactory::get()->CreateAudioInputIPC(render_frame_id));
 }
 
 // static

media/audio/audio_input_device.cc

@@ -85,16 +85,12 @@ class AudioInputDevice::AudioThreadCallback
   DISALLOW_COPY_AND_ASSIGN(AudioThreadCallback);
 };
 
-AudioInputDevice::AudioInputDevice(
-    std::unique_ptr<AudioInputIPC> ipc,
-    const scoped_refptr<base::SingleThreadTaskRunner>& io_task_runner)
-    : ScopedTaskRunnerObserver(io_task_runner),
-      callback_(NULL),
+AudioInputDevice::AudioInputDevice(std::unique_ptr<AudioInputIPC> ipc)
+    : callback_(nullptr),
       ipc_(std::move(ipc)),
       state_(IDLE),
       session_id_(0),
-      agc_is_enabled_(false),
-      stopping_hack_(false) {
+      agc_is_enabled_(false) {
   CHECK(ipc_);
 
   // The correctness of the code depends on the relative values assigned in the
@@ -107,14 +103,7 @@ AudioInputDevice::AudioInputDevice(
 void AudioInputDevice::Initialize(const AudioParameters& params,
                                   CaptureCallback* callback,
                                   int session_id) {
-  task_runner()->PostTask(
-      FROM_HERE, base::BindOnce(&AudioInputDevice::InitializeOnIOThread, this,
-                                params, callback, session_id));
-}
-
-void AudioInputDevice::InitializeOnIOThread(const AudioParameters& params,
-                                            CaptureCallback* callback,
-                                            int session_id) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   DCHECK(params.IsValid());
   DCHECK(!callback_);
   DCHECK_EQ(0, session_id_);
@@ -124,25 +113,58 @@ void AudioInputDevice::InitializeOnIOThread(const AudioParameters& params,
 }
 
 void AudioInputDevice::Start() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  DCHECK(callback_) << "Initialize hasn't been called";
   TRACE_EVENT0("audio", "AudioInputDevice::Start");
-  task_runner()->PostTask(
-      FROM_HERE, base::BindOnce(&AudioInputDevice::StartUpOnIOThread, this));
+
+  // Make sure we don't call Start() more than once.
+  if (state_ != IDLE)
+    return;
+
+  if (session_id_ <= 0) {
+    DLOG(WARNING) << "Invalid session id for the input stream " << session_id_;
+    return;
+  }
+
+  state_ = CREATING_STREAM;
+  ipc_->CreateStream(this, session_id_, audio_parameters_, agc_is_enabled_,
+                     kRequestedSharedMemoryCount);
 }
 
 void AudioInputDevice::Stop() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   TRACE_EVENT0("audio", "AudioInputDevice::Stop");
 
-  {
-    base::AutoLock auto_lock(audio_thread_lock_);
   audio_thread_.reset();
-    stopping_hack_ = true;
+
+  UMA_HISTOGRAM_BOOLEAN(
+      "Media.Audio.Capture.DetectedMissingCallbacks",
+      alive_checker_ ? alive_checker_->DetectedDead() : false);
+
+  UMA_HISTOGRAM_BOOLEAN("Media.Audio.Capture.StreamCallbackError",
+                        had_callback_error_);
+
+  // Close the stream, if we haven't already.
+  if (state_ >= CREATING_STREAM) {
+    ipc_->CloseStream();
+    state_ = IDLE;
+    agc_is_enabled_ = false;
   }
 
-  task_runner()->PostTask(
-      FROM_HERE, base::BindOnce(&AudioInputDevice::ShutDownOnIOThread, this));
+  // We can run into an issue where Stop is called right after
+  // OnStreamCreated is called in cases where Start/Stop are called before we
+  // get the OnStreamCreated callback.  To handle that corner case, we call
+  // audio_tread.reset(). In most cases, the thread will already be stopped.
+  //
+  // |alive_checker_| must outlive |audio_callback_|.
+  base::ScopedAllowBlocking allow_blocking;
+  audio_thread_.reset();
+  audio_callback_.reset();
+  alive_checker_.reset();
 }
 
 void AudioInputDevice::SetVolume(double volume) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   TRACE_EVENT1("audio", "AudioInputDevice::SetVolume", "volume", volume);
 
   if (volume < 0 || volume > 1.0) {
@@ -150,26 +172,30 @@ void AudioInputDevice::SetVolume(double volume) {
     return;
   }
 
-  task_runner()->PostTask(
-      FROM_HERE,
-      base::BindOnce(&AudioInputDevice::SetVolumeOnIOThread, this, volume));
+  if (state_ >= CREATING_STREAM)
+    ipc_->SetVolume(volume);
 }
 
 void AudioInputDevice::SetAutomaticGainControl(bool enabled) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   TRACE_EVENT1("audio", "AudioInputDevice::SetAutomaticGainControl", "enabled",
                enabled);
 
-  task_runner()->PostTask(
-      FROM_HERE,
-      base::BindOnce(&AudioInputDevice::SetAutomaticGainControlOnIOThread, this,
-                     enabled));
+  if (state_ >= CREATING_STREAM) {
+    DLOG(WARNING) << "The AGC state can not be modified after starting.";
+    return;
+  }
+
+  // We simply store the new AGC setting here. This value will be used when
+  // a new stream is initialized and by GetAutomaticGainControl().
+  agc_is_enabled_ = enabled;
 }
 
 void AudioInputDevice::OnStreamCreated(base::SharedMemoryHandle handle,
                                        base::SyncSocket::Handle socket_handle,
                                        bool initially_muted) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   TRACE_EVENT0("audio", "AudioInputDevice::OnStreamCreated");
-  DCHECK(task_runner()->BelongsToCurrentThread());
   DCHECK(base::SharedMemory::IsHandleValid(handle));
 #if defined(OS_WIN)
   DCHECK(socket_handle);
@@ -181,13 +207,6 @@ void AudioInputDevice::OnStreamCreated(base::SharedMemoryHandle handle,
   if (state_ != CREATING_STREAM)
     return;
 
-  base::AutoLock auto_lock(audio_thread_lock_);
-  // TODO(miu): See TODO in OnStreamCreated method for AudioOutputDevice.
-  // Interface changes need to be made; likely, after AudioInputDevice is merged
-  // into AudioOutputDevice (http://crbug.com/179597).
-  if (stopping_hack_)
-    return;
-
   DCHECK(!audio_callback_);
   DCHECK(!audio_thread_);
 
@@ -196,7 +215,7 @@ void AudioInputDevice::OnStreamCreated(base::SharedMemoryHandle handle,
 
 // Set up checker for detecting missing audio data. We pass a callback which
 // holds a reference to this. |alive_checker_| is deleted in
-// ShutDownOnIOThread() which we expect to always be called (see comment in
+// Stop() which we expect to always be called (see comment in
 // destructor). Suspend/resume notifications are not supported on Linux and
 // there's a risk of false positives when suspending. So on Linux we only detect
 // missing audio data until the first audio buffer arrives. Note that there's
@@ -232,8 +251,8 @@ void AudioInputDevice::OnStreamCreated(base::SharedMemoryHandle handle,
 }
 
 void AudioInputDevice::OnError() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   TRACE_EVENT0("audio", "AudioInputDevice::OnError");
-  DCHECK(task_runner()->BelongsToCurrentThread());
 
   // Do nothing if the stream has been closed.
   if (state_ < CREATING_STREAM)
@@ -257,15 +276,14 @@ void AudioInputDevice::OnError() {
     // TODO(tommi): Add an explicit contract for clearing the callback
     // object.  Possibly require calling Initialize again or provide
     // a callback object via Start() and clear it in Stop().
-    base::AutoLock auto_lock_(audio_thread_lock_);
     if (audio_thread_)
       callback_->OnCaptureError("IPC delegate state error.");
   }
 }
 
 void AudioInputDevice::OnMuted(bool is_muted) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   TRACE_EVENT0("audio", "AudioInputDevice::OnMuted");
-  DCHECK(task_runner()->BelongsToCurrentThread());
 
   // Do nothing if the stream has been closed.
   if (state_ < CREATING_STREAM)
@@ -274,8 +292,8 @@ void AudioInputDevice::OnMuted(bool is_muted) {
 }
 
 void AudioInputDevice::OnIPCClosed() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   TRACE_EVENT0("audio", "AudioInputDevice::OnIPCClosed");
-  DCHECK(task_runner()->BelongsToCurrentThread());
 
   state_ = IPC_CLOSED;
   ipc_.reset();
@@ -284,95 +302,14 @@ void AudioInputDevice::OnIPCClosed() {
 AudioInputDevice::~AudioInputDevice() {
 #if DCHECK_IS_ON()
   // Make sure we've stopped the stream properly before destructing |this|.
-  DCHECK(audio_thread_lock_.Try());
   DCHECK_LE(state_, IDLE);
   DCHECK(!audio_thread_);
   DCHECK(!audio_callback_);
   DCHECK(!alive_checker_);
-  DCHECK(!stopping_hack_);
-  audio_thread_lock_.Release();
 #endif  // DCHECK_IS_ON()
 }
 
-void AudioInputDevice::StartUpOnIOThread() {
-  DCHECK(task_runner()->BelongsToCurrentThread());
-  DCHECK(callback_) << "Initialize hasn't been called";
-
-  // Make sure we don't call Start() more than once.
-  if (state_ != IDLE)
-    return;
-
-  if (session_id_ <= 0) {
-    DLOG(WARNING) << "Invalid session id for the input stream " << session_id_;
-    return;
-  }
-
-  state_ = CREATING_STREAM;
-  ipc_->CreateStream(this, session_id_, audio_parameters_,
-                     agc_is_enabled_, kRequestedSharedMemoryCount);
-}
-
-void AudioInputDevice::ShutDownOnIOThread() {
-  DCHECK(task_runner()->BelongsToCurrentThread());
-
-  UMA_HISTOGRAM_BOOLEAN(
-      "Media.Audio.Capture.DetectedMissingCallbacks",
-      alive_checker_ ? alive_checker_->DetectedDead() : false);
-
-  UMA_HISTOGRAM_BOOLEAN("Media.Audio.Capture.StreamCallbackError",
-                        had_callback_error_);
-
-  // Close the stream, if we haven't already.
-  if (state_ >= CREATING_STREAM) {
-    ipc_->CloseStream();
-    state_ = IDLE;
-    agc_is_enabled_ = false;
-  }
-
-  // We can run into an issue where ShutDownOnIOThread is called right after
-  // OnStreamCreated is called in cases where Start/Stop are called before we
-  // get the OnStreamCreated callback.  To handle that corner case, we call
-  // Stop(). In most cases, the thread will already be stopped.
-  //
-  // Another situation is when the IO thread goes away before Stop() is called
-  // in which case, we cannot use the message loop to close the thread handle
-  // and can't not rely on the main thread existing either.
-  //
-  // |alive_checker_| must outlive |audio_callback_|.
-  base::AutoLock auto_lock_(audio_thread_lock_);
-  base::ThreadRestrictions::ScopedAllowIO allow_io;
-  audio_thread_.reset();
-  audio_callback_.reset();
-  alive_checker_.reset();
-  stopping_hack_ = false;
-}
-
-void AudioInputDevice::SetVolumeOnIOThread(double volume) {
-  DCHECK(task_runner()->BelongsToCurrentThread());
-  if (state_ >= CREATING_STREAM)
-    ipc_->SetVolume(volume);
-}
-
-void AudioInputDevice::SetAutomaticGainControlOnIOThread(bool enabled) {
-  DCHECK(task_runner()->BelongsToCurrentThread());
-
-  if (state_ >= CREATING_STREAM) {
-    DLOG(WARNING) << "The AGC state can not be modified after starting.";
-    return;
-  }
-
-  // We simply store the new AGC setting here. This value will be used when
-  // a new stream is initialized and by GetAutomaticGainControl().
-  agc_is_enabled_ = enabled;
-}
-
-void AudioInputDevice::WillDestroyCurrentMessageLoop() {
-  LOG(ERROR) << "IO loop going away before the input device has been stopped";
-  ShutDownOnIOThread();
-}
-
 void AudioInputDevice::DetectedDeadInputStream() {
-  DCHECK(task_runner()->BelongsToCurrentThread());
   callback_->OnCaptureError("No audio received from audio capture device.");
 }
 

media/audio/audio_input_device.h

@@ -27,25 +27,17 @@
 //
 // State sequences:
 //
-// Start -> InitializeOnIOThread -> CreateStream ->
+// Start -> CreateStream ->
 //       <- OnStreamCreated <-
-//       -> StartOnIOThread -> PlayStream ->
-//
+//       -> RecordStream ->
 //
 // AudioInputDevice::Capture => low latency audio transport on audio thread =>
-//                               |
-// Stop --> ShutDownOnIOThread ------>  CloseStream -> Close
 //
-// This class depends on two threads to function:
+// Stop ->  CloseStream -> Close
 //
-// 1. An IO thread.
-//    This thread is used to asynchronously process Start/Stop etc operations
-//    that are available via the public interface.  The public methods are
-//    asynchronous and simply post a task to the IO thread to actually perform
-//    the work.
-// 2. Audio transport thread.
-//    Responsible for calling the CaptureCallback and feed audio samples from
-//    the server side audio layer using a socket and shared memory.
+// This class depends on the audio transport thread. That thread is responsible
+// for calling the CaptureCallback and feeding it audio samples from the server
+// side audio layer using a socket and shared memory.
 //
 // Implementation notes:
 // - The user must call Stop() before deleting the class instance.
@@ -59,32 +51,22 @@
 #include "base/compiler_specific.h"
 #include "base/macros.h"
 #include "base/memory/shared_memory.h"
+#include "base/sequence_checker.h"
 #include "base/time/time.h"
-#include "base/timer/timer.h"
 #include "media/audio/alive_checker.h"
 #include "media/audio/audio_device_thread.h"
 #include "media/audio/audio_input_ipc.h"
-#include "media/audio/scoped_task_runner_observer.h"
 #include "media/base/audio_capturer_source.h"
 #include "media/base/audio_parameters.h"
 #include "media/base/media_export.h"
 
 namespace media {
 
-// TODO(henrika): This class is based on the AudioOutputDevice class and it has
-// many components in common. Investigate potential for re-factoring.
-// See http://crbug.com/179597.
-// TODO(henrika): Add support for event handling (e.g. OnStateChanged,
-// OnCaptureStopped etc.) and ensure that we can deliver these notifications
-// to any clients using this class.
 class MEDIA_EXPORT AudioInputDevice : public AudioCapturerSource,
-                                      public AudioInputIPCDelegate,
-                                      public ScopedTaskRunnerObserver {
+                                      public AudioInputIPCDelegate {
  public:
   // NOTE: Clients must call Initialize() before using.
-  AudioInputDevice(
-      std::unique_ptr<AudioInputIPC> ipc,
-      const scoped_refptr<base::SingleThreadTaskRunner>& io_task_runner);
+  AudioInputDevice(std::unique_ptr<AudioInputIPC> ipc);
 
   // AudioCapturerSource implementation.
   void Initialize(const AudioParameters& params,
@@ -111,7 +93,6 @@ class MEDIA_EXPORT AudioInputDevice : public AudioCapturerSource,
 
   ~AudioInputDevice() override;
 
-  // Methods called on IO thread ----------------------------------------------
   // AudioInputIPCDelegate implementation.
   void OnStreamCreated(base::SharedMemoryHandle handle,
                        base::SyncSocket::Handle socket_handle,
@@ -120,24 +101,8 @@ class MEDIA_EXPORT AudioInputDevice : public AudioCapturerSource,
   void OnMuted(bool is_muted) override;
   void OnIPCClosed() override;
 
-  // Methods called on IO thread ----------------------------------------------
-  // The following methods are tasks posted on the IO thread that needs to
-  // be executed on that thread.
-  void InitializeOnIOThread(const AudioParameters& params,
-                            CaptureCallback* callback,
-                            int session_id);
-  void StartUpOnIOThread();
-  void ShutDownOnIOThread();
-  void SetVolumeOnIOThread(double volume);
-  void SetAutomaticGainControlOnIOThread(bool enabled);
-
-  // base::MessageLoopCurrent::DestructionObserver implementation for the IO
-  // loop. If the IO loop dies before we do, we shut down the audio thread from
-  // here.
-  void WillDestroyCurrentMessageLoop() override;
-
   // This is called by |alive_checker_| if it detects that the input stream is
-  // dead. Called on the IO thread.
+  // dead.
   void DetectedDeadInputStream();
 
   AudioParameters audio_parameters_;
@@ -145,46 +110,30 @@ class MEDIA_EXPORT AudioInputDevice : public AudioCapturerSource,
   CaptureCallback* callback_;
 
   // A pointer to the IPC layer that takes care of sending requests over to
-  // the stream implementation.  Only valid when state_ != IPC_CLOSED and must
-  // only be accessed on the IO thread.
+  // the stream implementation.  Only valid when state_ != IPC_CLOSED.
   std::unique_ptr<AudioInputIPC> ipc_;
 
-  // Current state (must only be accessed from the IO thread).  See comments for
-  // State enum above.
+  // Current state. See comments for State enum above.
   State state_;
 
-  // For UMA stats. May only be accessed on the IO thread.
+  // For UMA stats.
   bool had_callback_error_ = false;
 
   // The media session ID used to identify which input device to be started.
-  // Only modified in Initialize() and ShutDownOnIOThread().
+  // Only modified in Initialize().
   int session_id_;
 
   // Stores the Automatic Gain Control state. Default is false.
-  // Only modified on the IO thread.
   bool agc_is_enabled_;
 
-  // In order to avoid a race between OnStreamCreated and Stop(), we use this
-  // guard to control stopping and starting the audio thread.
-  base::Lock audio_thread_lock_;
-
   // Checks regularly that the input stream is alive and notifies us if it
-  // isn't by calling DetectedDeadInputStream(). Created and deleted on the IO
-  // thread. Must outlive |audio_callback_|.
+  // isn't by calling DetectedDeadInputStream(). Must outlive |audio_callback_|.
   std::unique_ptr<AliveChecker> alive_checker_;
 
-  // Created and deleted on the IO thread, with the exception of in Stop(),
-  // where |audio_thread_| is reset (see comment on |audio_thread_lock_| above).
   std::unique_ptr<AudioInputDevice::AudioThreadCallback> audio_callback_;
   std::unique_ptr<AudioDeviceThread> audio_thread_;
 
-  // Temporary hack to ignore OnStreamCreated() due to the user calling Stop()
-  // so we don't start the audio thread pointing to a potentially freed
-  // |callback_|.
-  //
-  // TODO(miu): Replace this by changing AudioCapturerSource to accept the
-  // callback via Start(). See http://crbug.com/151051 for details.
-  bool stopping_hack_;
+  SEQUENCE_CHECKER(sequence_checker_);
 
   DISALLOW_IMPLICIT_CONSTRUCTORS(AudioInputDevice);
 };

media/audio/audio_input_device_unittest.cc

@@ -2,6 +2,7 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include "media/audio/audio_input_device.h"
 #include "base/memory/ptr_util.h"
 #include "base/memory/shared_memory.h"
 #include "base/message_loop/message_loop.h"
@@ -9,7 +10,7 @@
 #include "base/run_loop.h"
 #include "base/single_thread_task_runner.h"
 #include "base/sync_socket.h"
-#include "media/audio/audio_input_device.h"
+#include "base/test/scoped_task_environment.h"
 #include "testing/gmock/include/gmock/gmock.h"
 #include "testing/gmock_mutant.h"
 #include "testing/gtest/include/gtest/gtest.h"
@@ -59,12 +60,6 @@ class MockCaptureCallback : public AudioCapturerSource::CaptureCallback {
   MOCK_METHOD1(OnCaptureMuted, void(bool is_muted));
 };
 
-// Used to terminate a loop from a different thread than the loop belongs to.
-// |task_runner| should be a SingleThreadTaskRunner.
-ACTION_P(QuitLoop, task_runner) {
-  task_runner->PostTask(FROM_HERE, base::MessageLoop::QuitWhenIdleClosure());
-}
-
 }  // namespace.
 
 // Regular construction.
@@ -72,7 +67,7 @@ TEST(AudioInputDeviceTest, Noop) {
   base::MessageLoopForIO io_loop;
   MockAudioInputIPC* input_ipc = new MockAudioInputIPC();
   scoped_refptr<AudioInputDevice> device(
-      new AudioInputDevice(base::WrapUnique(input_ipc), io_loop.task_runner()));
+      new AudioInputDevice(base::WrapUnique(input_ipc)));
 }
 
 ACTION_P(ReportStateChange, device) {
@@ -84,20 +79,16 @@ TEST(AudioInputDeviceTest, FailToCreateStream) {
   AudioParameters params(AudioParameters::AUDIO_PCM_LOW_LATENCY,
                          CHANNEL_LAYOUT_STEREO, 48000, 16, 480);
 
-  base::MessageLoopForIO io_loop;
   MockCaptureCallback callback;
   MockAudioInputIPC* input_ipc = new MockAudioInputIPC();
   scoped_refptr<AudioInputDevice> device(
-      new AudioInputDevice(base::WrapUnique(input_ipc), io_loop.task_runner()));
+      new AudioInputDevice(base::WrapUnique(input_ipc)));
   device->Initialize(params, &callback, 1);
-  device->Start();
   EXPECT_CALL(*input_ipc, CreateStream(_, _, _, _, _))
       .WillOnce(ReportStateChange(device.get()));
-  EXPECT_CALL(callback, OnCaptureError(_))
-      .WillOnce(QuitLoop(io_loop.task_runner()));
-  base::RunLoop().Run();
+  EXPECT_CALL(callback, OnCaptureError(_));
+  device->Start();
   device->Stop();
-  base::RunLoop().RunUntilIdle();
 }
 
 ACTION_P3(ReportOnStreamCreated, device, handle, socket) {
@@ -127,24 +118,23 @@ TEST(AudioInputDeviceTest, CreateStream) {
       shared_memory.handle().Duplicate();
   ASSERT_TRUE(duplicated_memory_handle.IsValid());
 
-  base::MessageLoopForIO io_loop;
+  base::test::ScopedTaskEnvironment ste;
   MockCaptureCallback callback;
   MockAudioInputIPC* input_ipc = new MockAudioInputIPC();
   scoped_refptr<AudioInputDevice> device(
-      new AudioInputDevice(base::WrapUnique(input_ipc), io_loop.task_runner()));
+      new AudioInputDevice(base::WrapUnique(input_ipc)));
   device->Initialize(params, &callback, 1);
-  device->Start();
 
   EXPECT_CALL(*input_ipc, CreateStream(_, _, _, _, _))
       .WillOnce(ReportOnStreamCreated(
           device.get(), duplicated_memory_handle,
           SyncSocket::UnwrapHandle(audio_device_socket_descriptor)));
   EXPECT_CALL(*input_ipc, RecordStream());
-  EXPECT_CALL(callback, OnCaptureStarted())
-      .WillOnce(QuitLoop(io_loop.task_runner()));
-  base::RunLoop().Run();
+
+  EXPECT_CALL(callback, OnCaptureStarted());
+  device->Start();
+  EXPECT_CALL(*input_ipc, CloseStream());
   device->Stop();
-  base::RunLoop().RunUntilIdle();
   duplicated_memory_handle.Close();
 }