Make AudioOutputDevice restartable and reinitializable

media/audio/audio_device_thread.cc

 // Copyright (c) 2012 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 "media/audio/audio_device_thread.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <algorithm>
 #include <limits>
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/memory/aligned_memory.h"
 #include "base/message_loop/message_loop.h"
 #include "base/numerics/safe_conversions.h"
+#include "base/synchronization/condition_variable.h"
 #include "base/threading/platform_thread.h"
+#include "base/threading/thread_checker.h"
 #include "base/threading/thread_restrictions.h"
 #include "media/base/audio_bus.h"
 
 using base::PlatformThread;
 
 namespace media {
 
 // The actual worker thread implementation.  It's very bare bones and much
 // simpler than SimpleThread (no synchronization in Start, etc) and supports
 // joining the thread handle asynchronously via a provided message loop even
 // after the Thread object itself has been deleted.
 class AudioDeviceThread::Thread
     : public PlatformThread::Delegate,
       public base::RefCountedThreadSafe<AudioDeviceThread::Thread> {
  public:
-  Thread(AudioDeviceThread::Callback* callback,
-         base::SyncSocket::Handle socket,
-         const char* thread_name,
-         bool synchronized_buffers);
-
-  void Start();
-
-  // Stops the thread.  If |loop_for_join| is non-NULL, the function posts
+  Thread(const char* thread_name, bool synchronized_buffers);
+
+  // Starts the thread and suspends it, which means it will wait in
+  // ThreadMain(). Resume() must then be called to actually start running it.
+  void StartSuspended();
+
+  // Shuts down the thread.  If |loop_for_join| is non-NULL, the function posts
   // a task to join (close) the thread handle later instead of waiting for
   // the thread.  If loop_for_join is NULL, then the function waits
   // synchronously for the thread to terminate.
+  // Must be called before destruction.
   void Stop(base::MessageLoop* loop_for_join);
 
+  // Resumes the thread and starts running it with the given |callback| and
+  // |socket|.
+  void Resume(AudioDeviceThread::Callback* callback,
+              base::SyncSocket::Handle socket);
+
+  // Suspends the thread, it will wait in ThreadMain(). Shuts down the socket.
+  // It can be resumed again by calling Resume().
+  void Suspend();
+
  private:
+  enum State { STATE_SUSPENDED, STATE_PLAYING, STATE_STOPPED };

[DaleCurtis, 2016/05/17 17:54:35]

Instead of adding STATE_ consider making this an "enum class" Also this should
live down near the actual state_ variable.

[Henrik Grunell, 2016/05/18 12:21:32]

Good points, though can't use DCHECK_EQ etc. with enum class types. Both done.

+
   friend class base::RefCountedThreadSafe<AudioDeviceThread::Thread>;

[DaleCurtis, 2016/05/17 17:54:35]

friend goes right under private:

[Henrik Grunell, 2016/05/18 12:21:32]

Done.

+
   ~Thread() override;
 
   // Overrides from PlatformThread::Delegate.
   void ThreadMain() override;
 
-  // Runs the loop that reads from the socket.
+  // Runs the loop that reads from the socket, called in ThreadMain().
   void Run();
 
- private:
-  base::PlatformThreadHandle thread_;
-  AudioDeviceThread::Callback* callback_;
-  base::CancelableSyncSocket socket_;
-  base::Lock callback_lock_;
+  // Ensures that Start(), Resume() and Suspend() are called on the same thread.
+  // Stop() can be called on any thread.
+  base::ThreadChecker thread_checker_;
+
   const char* thread_name_;
   const bool synchronized_buffers_;
 
+  base::Lock lock_;  // Protects everything below.
+  base::ConditionVariable exit_suspended_state_;
+  State state_;
+  AudioDeviceThread::Callback* callback_;
+
+  base::PlatformThreadHandle thread_handle_;
+
+  // Reset in ThreadMain(). Shutdown on another thread.
+  std::unique_ptr<base::CancelableSyncSocket> socket_;
+
+  base::SyncSocket::Handle new_socket_handle_;
+
   DISALLOW_COPY_AND_ASSIGN(Thread);
 };
 
 // AudioDeviceThread implementation
-
-AudioDeviceThread::AudioDeviceThread() {
-}
-
-AudioDeviceThread::~AudioDeviceThread() { DCHECK(!thread_.get()); }
-
-void AudioDeviceThread::Start(AudioDeviceThread::Callback* callback,
-                              base::SyncSocket::Handle socket,
-                              const char* thread_name,
-                              bool synchronized_buffers) {
+AudioDeviceThread::AudioDeviceThread() {}
+
+AudioDeviceThread::~AudioDeviceThread() {
+  DCHECK(!thread_.get());
+}
+
+void AudioDeviceThread::StartSuspended(const char* thread_name,
+                                       bool synchronized_buffers) {
   base::AutoLock auto_lock(thread_lock_);
   CHECK(!thread_.get());
-  thread_ = new AudioDeviceThread::Thread(
-      callback, socket, thread_name, synchronized_buffers);
-  thread_->Start();
+  thread_ = new AudioDeviceThread::Thread(thread_name, synchronized_buffers);
+  thread_->StartSuspended();
 }
 
 void AudioDeviceThread::Stop(base::MessageLoop* loop_for_join) {
   base::AutoLock auto_lock(thread_lock_);
   if (thread_.get()) {
     thread_->Stop(loop_for_join);
-    thread_ = NULL;
+    thread_ = nullptr;
   }
 }
 
+void AudioDeviceThread::Resume(AudioDeviceThread::Callback* callback,
+                               base::SyncSocket::Handle socket) {
+  base::AutoLock auto_lock(thread_lock_);
+  if (thread_.get())

[DaleCurtis, 2016/05/17 17:54:35]

Drop .get() here and elsewhere.

[Henrik Grunell, 2016/05/18 12:21:32]

Done.

+    thread_->Resume(callback, socket);
+}
+
+void AudioDeviceThread::Suspend() {
+  base::AutoLock auto_lock(thread_lock_);
+  if (thread_.get())
+    thread_->Suspend();
+}
+
 bool AudioDeviceThread::IsStopped() {
   base::AutoLock auto_lock(thread_lock_);
   return !thread_.get();
 }
 
 // AudioDeviceThread::Thread implementation
-AudioDeviceThread::Thread::Thread(AudioDeviceThread::Callback* callback,
-                                  base::SyncSocket::Handle socket,
-                                  const char* thread_name,
+AudioDeviceThread::Thread::Thread(const char* thread_name,
                                   bool synchronized_buffers)
-    : thread_(),
-      callback_(callback),
-      socket_(socket),
-      thread_name_(thread_name),
-      synchronized_buffers_(synchronized_buffers) {
-}
+    : thread_name_(thread_name),
+      synchronized_buffers_(synchronized_buffers),
+      exit_suspended_state_(&lock_),
+      state_(STATE_SUSPENDED),
+      callback_(nullptr),
+      new_socket_handle_(base::CancelableSyncSocket::kInvalidHandle) {}
 
 AudioDeviceThread::Thread::~Thread() {
-  DCHECK(thread_.is_null());
-}
-
-void AudioDeviceThread::Thread::Start() {
-  base::AutoLock auto_lock(callback_lock_);
-  DCHECK(thread_.is_null());
-  // This reference will be released when the thread exists.
+  DCHECK(thread_handle_.is_null());
+  DCHECK_EQ(state_, STATE_STOPPED)
+      << "Thread must be stopped before destruction";
+}
+
+void AudioDeviceThread::Thread::StartSuspended() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  base::AutoLock auto_lock(lock_);
+  DCHECK(thread_handle_.is_null()) << "Calling Start() on a started thread?";
+  if (state_ == STATE_STOPPED)
+    return;  // Stop has already been executed on another thread.
+
+  // This reference will be released when the thread exits.

[DaleCurtis, 2016/05/17 17:54:35]

We should try to remove this in the future too; pretty sketchy.

[Henrik Grunell, 2016/05/18 12:21:32]

Agree, added TODO.

   AddRef();
 
-  PlatformThread::CreateWithPriority(0, this, &thread_,
+  PlatformThread::CreateWithPriority(0, this, &thread_handle_,
                                      base::ThreadPriority::REALTIME_AUDIO);
-  CHECK(!thread_.is_null());
+  CHECK(!thread_handle_.is_null());
 }
 
 void AudioDeviceThread::Thread::Stop(base::MessageLoop* loop_for_join) {
-  socket_.Shutdown();
-
-  base::PlatformThreadHandle thread = base::PlatformThreadHandle();
-
-  {  // NOLINT
-    base::AutoLock auto_lock(callback_lock_);
-    callback_ = NULL;
-    std::swap(thread, thread_);
-  }
-
-  if (!thread.is_null()) {
+  // Note: Stop() can be called at any time, even before Start(), because it's
+  // allowed to be executed on a different thread.
+
+  base::PlatformThreadHandle thread_handle_to_join =
+      base::PlatformThreadHandle();
+  {
+    base::AutoLock auto_lock(lock_);
+    DCHECK_NE(state_, STATE_STOPPED) << "Calling Stop() on a stopped thread?";
+    callback_ = nullptr;
+
+    // |socket_| is only set/reset under the |lock_|, so it's safe to
+    // access under that lock here.
+    if (socket_)
+      socket_->Shutdown();
+
+    // Must be done under lock to avoid race with Start().
+    std::swap(thread_handle_to_join, thread_handle_);
+
+    state_ = STATE_STOPPED;
+    exit_suspended_state_.Signal();
+  }  // auto_lock(lock_)
+
+  if (!thread_handle_to_join.is_null()) {
     if (loop_for_join) {
-      loop_for_join->PostTask(FROM_HERE,
-          base::Bind(&base::PlatformThread::Join, thread));
+      loop_for_join->PostTask(FROM_HERE, base::Bind(&base::PlatformThread::Join,
+                                                    thread_handle_to_join));
     } else {
-      base::PlatformThread::Join(thread);
+      base::PlatformThread::Join(thread_handle_to_join);
     }
   }
 }
 
+void AudioDeviceThread::Thread::Resume(AudioDeviceThread::Callback* callback,
+                                       base::SyncSocket::Handle socket_handle) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  base::AutoLock auto_lock(lock_);
+  if (state_ == STATE_STOPPED)
+    return;  // Stop is the last operation to be executed.
+
+  DCHECK_EQ(state_, STATE_SUSPENDED) << "Resume a non-suspended thread?";
+
+  DCHECK_EQ(new_socket_handle_, base::CancelableSyncSocket::kInvalidHandle);
+
+  callback_ = callback;
+  new_socket_handle_ = socket_handle;
+
+  state_ = STATE_PLAYING;
+  exit_suspended_state_.Signal();
+}
+
+void AudioDeviceThread::Thread::Suspend() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  base::AutoLock auto_lock(lock_);
+  if (state_ == STATE_STOPPED)
+    return;  // Stop is the last operation to be executed.
+
+  DCHECK_EQ(state_, STATE_PLAYING) << "Suspend a suspended thread?";
+  callback_ = nullptr;
+
+  // |socket_| is only set/reset under the |lock_|, so it's safe to
+  // access under that lock here.
+  if (socket_)
+    socket_->Shutdown();
+  state_ = STATE_SUSPENDED;
+}
+
 void AudioDeviceThread::Thread::ThreadMain() {
   PlatformThread::SetName(thread_name_);
 
   // Singleton access is safe from this thread as long as callback is non-NULL.
   // The callback is the only point where the thread calls out to 'unknown' code
   // that might touch singletons and the lifetime of the callback is controlled
   // by another thread on which singleton access is OK as well.
   base::ThreadRestrictions::SetSingletonAllowed(true);
 
-  {  // NOLINT
-    base::AutoLock auto_lock(callback_lock_);
-    if (callback_)
-      callback_->InitializeOnAudioThread();
-  }
+  while (true) {
+    {
+      base::AutoLock auto_lock(lock_);
+      while (state_ == STATE_SUSPENDED) {

[DaleCurtis, 2016/05/17 17:54:35]

What's the point of doing this versus just restarting the thread each time?

[Henrik Grunell, 2016/05/18 12:21:32]

The problem with spinning up a new thread each time is that in many places in
the code we assume render calls on the same thread and have checks for this.
It's complicated to inform/restart them to accept a new thread.

I put this in the CL description too.

[DaleCurtis, 2016/05/18 17:20:43]

I feel like overcomplicating our production code to workaround debug checks is
not a great course of action.  What do you think about introducing an
AudioRendererSink::CurrentlyOnAudioRenderingThread() type helper method that we
can replace all of these type checks with? It would simply be a tid comparison
implemented by the underlying device.

[Henrik Grunell, 2016/05/19 13:03:11]

The problems is more specifically the components used in clients in the render
callbacks in various places, also in blink iirc, that have thread checks. Those
components don't know of AudioRendererSinks or who uses them. I did write down
what dchecks we hit, but I couldn't find it now. So that option is unfortunately
not viable.

[DaleCurtis, 2016/05/19 20:41:00]

I don't think that's a good enough reason to make this change instead of fixing
those call sites; this adds quite a bit of extra complexity to an already
complex, high-frequency path to workaround some debug checks. Can you elaborate
on why the approach is not viable?

It doesn't seem like it should be all that difficult to change how those debug
checks work. Even though Blink doesn't know about sinks, it does know about
blink::WebAudioDevice which knows about media::AudioRendererSink, so it's seems
like it would just be a proxied call away.

[Henrik Grunell, 2016/05/23 11:41:58]

Actually, I have brought up this thread re-use approach for discussion before,
and you approved of it then. I'll wake up that mail thread and you can take a
look at that.

+        // This releases |lock_| while waiting.
+        exit_suspended_state_.Wait();
+      }
 
-  Run();
+      if (state_ == STATE_STOPPED)
+        break;
+
+      // If we are playing but don't have a valid new socket handle, we are here
+      // because the socket receive or send failed for another reason than
+      // Shutdown() being called in Stop() or Suspend(). In that case, quit the
+      // thread.
+      if (new_socket_handle_ == base::CancelableSyncSocket::kInvalidHandle) {
+        LOG(WARNING) << "Audio socket failed, exiting audio thread.";
+        break;
+      }
+
+      // Else reinitialize and resume.
+
+      socket_.reset(new base::CancelableSyncSocket(new_socket_handle_));
+
+      if (callback_)
+        callback_->InitializeOnAudioThread();
+
+      new_socket_handle_ = base::CancelableSyncSocket::kInvalidHandle;
+    }  // auto_lock(lock_)
+
+    Run();
+  }  // while
 
   // Release the reference for the thread. Note that after this, the Thread
   // instance will most likely be deleted.
   Release();
 }
 
 void AudioDeviceThread::Thread::Run() {
   uint32_t buffer_index = 0;
+
   while (true) {
     uint32_t pending_data = 0;
-    size_t bytes_read = socket_.Receive(&pending_data, sizeof(pending_data));
+    // |socket_| is only modified on this thread, so we don't need a lock to
+    // access it here.
+    size_t bytes_read = socket_->Receive(&pending_data, sizeof(pending_data));
+
     if (bytes_read != sizeof(pending_data))
       break;
 
     // std::numeric_limits<uint32_t>::max() is a special signal which is
     // returned after the browser stops the output device in response to a
     // renderer side request.
     //
     // Avoid running Process() for the paused signal, we still need to update
     // the buffer index if |synchronized_buffers_| is true though.
     //
     // See comments in AudioOutputController::DoPause() for details on why.
     if (pending_data != std::numeric_limits<uint32_t>::max()) {
-      base::AutoLock auto_lock(callback_lock_);
+      // We can't use an atomic pointer for |callback_|, but need to call on it
+      // under the state lock. This is because after clearing it in Suspend() or
+      // Stop(), the caller may delete the callback object.
+      base::AutoLock auto_lock(lock_);
       if (callback_)
         callback_->Process(pending_data);
     }
 
     // The usage of |synchronized_buffers_| differs between input and output
     // cases.
     // Input:
     // Let the other end know that we have read data, so that it can verify
     // it doesn't overwrite any data before read. The |buffer_index| value is
     // not used. For more details, see AudioInputSyncWriter::Write().
     // Output:
     // Let the other end know which buffer we just filled. The |buffer_index| is
     // used to ensure the other end is getting the buffer it expects. For more
     // details on how this works see AudioSyncReader::WaitUntilDataIsReady().
     if (synchronized_buffers_) {
       ++buffer_index;
-      size_t bytes_sent = socket_.Send(&buffer_index, sizeof(buffer_index));
+      size_t bytes_sent = socket_->Send(&buffer_index, sizeof(buffer_index));
+
       if (bytes_sent != sizeof(buffer_index))
         break;
     }
-  }
+  }  // while
 }
 
 // AudioDeviceThread::Callback implementation
 
 AudioDeviceThread::Callback::Callback(const AudioParameters& audio_parameters,
                                       base::SharedMemoryHandle memory,
                                       int memory_length,
                                       int total_segments)
     : audio_parameters_(audio_parameters),
       samples_per_ms_(static_cast<double>(audio_parameters.sample_rate()) /
@@ skipping 14 matching lines @@
 }
 
 AudioDeviceThread::Callback::~Callback() {}
 
 void AudioDeviceThread::Callback::InitializeOnAudioThread() {
   MapSharedMemory();
   CHECK(shared_memory_.memory());
 }
 
 }  // namespace media.