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_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);
+  Thread(const char* thread_name, bool synchronized_buffers);
 
+  // Starts the thread which will wait in ThreadMain(). Resume() must then be
+  // called to actually start running it.
   void Start();
 
-  // Stops the thread.  If |loop_for_join| is non-NULL, the function posts
+  // 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.

[o1ka, 2016/05/10 13:04:53]

.. before destruction? (It just sounds more complete to me, easier for brains :)
)

[Henrik Grunell, 2016/05/11 13:11:16]

Done.

   void Stop(base::MessageLoop* loop_for_join);
 
+  // Releases the thread from waiting and starts running it with the given
+  // |callback| and |socket|.
+  void Resume(AudioDeviceThread::Callback* callback,
+              base::SyncSocket::Handle socket);
+
+  // Pauses the thread, it will wait in ThreadMain(). Shuts down the
+  // socket. Resume() can then be called to restart it.
+  void Pause();
+
  private:
+  enum State { kStatePaused, kStatePlaying, kStateStopped };
+
   friend class base::RefCountedThreadSafe<AudioDeviceThread::Thread>;
+
   ~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 Pause() are called on the same thread.
+  // Stop() can be called any thread.

[o1ka, 2016/05/10 13:04:53]

nit: on/from any thread

[Henrik Grunell, 2016/05/11 13:11:16]

Done.

+  base::ThreadChecker thread_checker_;
+
   const char* thread_name_;
   const bool synchronized_buffers_;
 
+  base::Lock state_lock_;  // Protects everything below.
+  base::ConditionVariable state_condition_;
+  State state_;
+  AudioDeviceThread::Callback* callback_;
+
+  base::PlatformThreadHandle thread_;
+
+  // Reset in ThreadMain(). Shutdown on client thread.
+  std::unique_ptr<base::CancelableSyncSocket> socket_;
+
+  // Set on client thead.
+  base::SyncSocket::Handle new_socket_handle_;
+
   DISALLOW_COPY_AND_ASSIGN(Thread);
 };
 
-// AudioDeviceThread implementation

[o1ka, 2016/05/10 13:04:53]

Put the comment back?

[Henrik Grunell, 2016/05/11 13:11:16]

Done. (Although I don't think it or the one for AudioDeviceThread::Thread
implementation add readability. But I'm OK with either.)

+AudioDeviceThread::AudioDeviceThread() {}
 
-AudioDeviceThread::AudioDeviceThread() {
+AudioDeviceThread::~AudioDeviceThread() {
+  DCHECK(!thread_.get());
 }
 
-AudioDeviceThread::~AudioDeviceThread() { DCHECK(!thread_.get()); }
-
-void AudioDeviceThread::Start(AudioDeviceThread::Callback* callback,
-                              base::SyncSocket::Handle socket,
-                              const char* thread_name,
+void AudioDeviceThread::Start(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_ = new AudioDeviceThread::Thread(thread_name, synchronized_buffers);
   thread_->Start();
 }
 
 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())
+    thread_->Resume(callback, socket);
+}
+
+void AudioDeviceThread::Pause() {
+  base::AutoLock auto_lock(thread_lock_);
+  if (thread_.get())
+    thread_->Pause();
+}
+
 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),
+      state_condition_(&state_lock_),
+      state_(kStatePaused),
+      callback_(nullptr),
+      new_socket_handle_(base::CancelableSyncSocket::kInvalidHandle) {
+  thread_checker_.DetachFromThread();
 }
 
 AudioDeviceThread::Thread::~Thread() {
   DCHECK(thread_.is_null());
+  DCHECK_EQ(state_, kStateStopped);

[o1ka, 2016/05/10 13:04:53]

nit: DCHECK_EQ(state_, kStateStopped) << "Thread must be stopped before
destruction";
(a bit more verbal)

[Henrik Grunell, 2016/05/11 13:11:16]

Done.

 }
 
 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_checker_.CalledOnValidThread());
+
+  base::AutoLock auto_lock(state_lock_);
+  DCHECK(thread_.is_null()) << "Calling Start() on a started thread?";
+  if (state_ == kStateStopped)
+    return;  // Stop has already been executed on another thread.
+
+  // This reference will be released when the thread exits.
   AddRef();
 
   PlatformThread::CreateWithPriority(0, this, &thread_,
                                      base::ThreadPriority::REALTIME_AUDIO);
   CHECK(!thread_.is_null());
 }
 
 void AudioDeviceThread::Thread::Stop(base::MessageLoop* loop_for_join) {
-  socket_.Shutdown();
+  // Note: Stop() can be called any time, even before Start(), because it's
+  // allowed to be executed on a different thread.
 
   base::PlatformThreadHandle thread = base::PlatformThreadHandle();
+  {
+    base::AutoLock auto_lock(state_lock_);
+    DCHECK(state_ != kStateStopped) << "Calling Stop() on a stopped thread?";

[o1ka, 2016/05/10 13:04:53]

Why DCHECK? Can it be a problem? I think we allow to stop a device multiple
times, all Stop() calls but one are just no-ops.
(We can add DCHECKs there that if kStateStopped, then everything is stopped
properly though)

[Henrik Grunell, 2016/05/11 13:11:16]

Multiple calls to AudioDeviceThread::Stop() is allowed. But it will only call
AudioDeviceThread::Thread::Stop() the first time, then AudioDeviceThread::Thread
will be deleted. So the DCHECK is a correct assertion.

+    callback_ = nullptr;
 
-  {  // NOLINT
-    base::AutoLock auto_lock(callback_lock_);
-    callback_ = NULL;
+    // |socket_| is only set/reset under the |state_lock_|, so it's safe to
+    // access under that lock here.
+    if (socket_)
+      socket_->Shutdown();
+
+    // Must be done under lock to avoid race wuth Start().

[o1ka, 2016/05/10 13:04:53]

nit: wuth -> with

[Henrik Grunell, 2016/05/11 13:11:16]

Done.

     std::swap(thread, thread_);
-  }
+
+    state_ = kStateStopped;
+    state_condition_.Signal();
+  }  // auto_lock(state_lock_)
 
   if (!thread.is_null()) {
     if (loop_for_join) {
       loop_for_join->PostTask(FROM_HERE,
           base::Bind(&base::PlatformThread::Join, thread));
     } else {
       base::PlatformThread::Join(thread);
     }
   }
 }
 
+void AudioDeviceThread::Thread::Resume(AudioDeviceThread::Callback* callback,
+                                       base::SyncSocket::Handle socket_handle) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  base::AutoLock auto_lock(state_lock_);
+  if (state_ == kStateStopped)
+    return;  // Stop is the last operation to be executed.
+
+  DCHECK(state_ == kStatePaused) << "Resume a non-paused thread?";
+
+  DCHECK_EQ(new_socket_handle_, base::CancelableSyncSocket::kInvalidHandle);
+
+  callback_ = callback;
+  new_socket_handle_ = socket_handle;
+
+  state_ = kStatePlaying;
+  state_condition_.Signal();
+}
+
+void AudioDeviceThread::Thread::Pause() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  base::AutoLock auto_lock(state_lock_);
+  if (state_ == kStateStopped)
+    return;  // Stop is the last operation to be executed.
+
+  DCHECK(state_ == kStatePlaying) << "Pause a paused thread?";
+  callback_ = nullptr;
+
+  // |socket_| is only set/reset under the |state_lock_|, so it's safe to
+  // access under that lock here.
+  if (socket_)
+    socket_->Shutdown();
+  state_ = kStatePaused;
+}
+
 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(state_lock_);
+      while (state_ == kStatePaused)
+        state_condition_.Wait();
 
-  Run();
+      if (state_ == kStateStopped)
+        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(state_lock_)
+
+    Run();
+
+    // We are here if and only if socket operation failed. The possible reasons
+    // are:
+    // 1. Shutdown() has been called on the socket, i.e. Stop() or Pause() has
+    //    been called, or
+    // 2. Socket failed.
+    // In both cases go back and wait for stop or resume.
+  }  // 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 in ThreadMain, so it is safe to access 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_);
+      base::AutoLock auto_lock(state_lock_);

[o1ka, 2016/05/10 13:04:53]

Could you add a comment why we should hold a lock during render call and cannot
use atomic pointer here instead?

[Henrik Grunell, 2016/05/11 13:11:16]

Done.

       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.