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(). Play() 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.
   void Stop(base::MessageLoop* loop_for_join);
 
+  // Releases the thread from waiting and starts running it with the given
+  // |callback| and |socket|.
+  void Play(AudioDeviceThread::Callback* callback,
+            base::SyncSocket::Handle socket);
+
+  // Pauses the thread, it will wait in ThreadMain(). Shuts down the
+  // socket. Play() can then be called to restart it.
+  void Pause();
+
  private:
+  enum ResumeAction { kActionPause, kActionPlay, kActionStop };
+
   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_;
+  // Start(), Play(), Pause() must be serialized; Stop() can be called any
+  // moment.
+  base::ThreadChecker thread_checker_;
+
   const char* thread_name_;
   const bool synchronized_buffers_;
 
+  // Modified in ThreadMain only, no lock protection needed.
+  AudioDeviceThread::Callback* callback_;
+
+  base::Lock action_lock_;  // Protects everything below.
+  base::ConditionVariable action_condition_;
+  ResumeAction action_;
+
+  base::PlatformThreadHandle thread_;
+
+  // Reset on client threaid.
+  std::unique_ptr<base::CancelableSyncSocket> socket_;
+
+  // Set on client thead.
+  AudioDeviceThread::Callback* new_callback_;
+  base::SyncSocket::Handle new_socket_handle_;
+
   DISALLOW_COPY_AND_ASSIGN(Thread);
 };
 
-// AudioDeviceThread implementation
+// AudioDeviceThread implementation TODO(olka): get rid of it. It just adds one
+// extra lock.
 
-AudioDeviceThread::AudioDeviceThread() {
-}
+AudioDeviceThread::AudioDeviceThread() {}
 
 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;
   }
 }
 
+void AudioDeviceThread::Play(AudioDeviceThread::Callback* callback,
+                             base::SyncSocket::Handle socket) {
+  base::AutoLock auto_lock(thread_lock_);
+  thread_->Play(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),
+      callback_(nullptr),
+      action_condition_(&action_lock_),
+      action_(kActionPause),
+      new_callback_(nullptr),
+      new_socket_handle_(base::CancelableSyncSocket::kInvalidHandle) {
+  thread_checker_.DetachFromThread();
 }
 
 AudioDeviceThread::Thread::~Thread() {
   DCHECK(thread_.is_null());
 }
 
 void AudioDeviceThread::Thread::Start() {
-  base::AutoLock auto_lock(callback_lock_);
-  DCHECK(thread_.is_null());
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  base::AutoLock auto_lock(action_lock_);
+  DCHECK(thread_.is_null()) << "Calling Start() on a started thread?";
+  if (action_ == kActionStop)
+    return;  // Stop has already been executed on another thread.
+
   // This reference will be released when the thread exists.
   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();
+  // NB: 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(action_lock_);
+    DCHECK(action_ != kActionStop) << "Calling Stop() on a stopped thread?";
 
-  {  // NOLINT
-    base::AutoLock auto_lock(callback_lock_);
-    callback_ = NULL;
+    // Is it safe to shutdown socket twice or shutdown a socket with an invalid
+    // handle? If not, shutdown only if kActionPlay?
+    socket_->Shutdown();
+
+    // Must be done under lock to avoid race wuth Start().
     std::swap(thread, thread_);
-  }
+
+    action_ = kActionStop;
+    action_condition_.Signal();
+  }  // auto_lock(action_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::Play(AudioDeviceThread::Callback* callback,
+                                     base::SyncSocket::Handle socket_handle) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  base::AutoLock auto_lock(action_lock_);
+  if (action_ == kActionStop)
+    return;  // Stop is the last operation to be executed.
+
+  DCHECK(action_ != kActionPlay) << "Play a playing thread?";
+
+  DCHECK_EQ(new_socket_handle_, base::CancelableSyncSocket::kInvalidHandle);
+
+  new_callback_ = callback;
+  new_socket_handle_ = socket_handle;
+
+  action_ = kActionPlay;
+  action_condition_.Signal();
+}
+
+void AudioDeviceThread::Thread::Pause() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  base::AutoLock auto_lock(action_lock_);
+  if (action_ == kActionStop)
+    return;  // Stop is the last operation to be executed.
+
+  DCHECK(action_ != kActionPause) << "Pause a paused thread?";
+
+  // Is it safe to shutdown socket twice? If not, shutdown only if kActionPlay?
+  socket_->Shutdown();
+  action_ = kActionPause;
+}
+
 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(action_lock_);
+      while (action_ == kActionPause)
+        action_condition_.Wait();
+
+      if (action_ == kActionStop)
+        break;
+      // else reinitialize and resume.
+
+      socket_.reset(new base::CancelableSyncSocket(new_socket_handle_));
+
+      callback_ = new_callback_;
+      if (callback_)
+        callback_->InitializeOnAudioThread();
+
+      new_callback_ = nullptr;
+      new_socket_handle_ = base::CancelableSyncSocket::kInvalidHandle;
+    }  // auto_lock(action_lock_)
+
+    Run();
+
+    // We are here if and only if socket operation failed. The possible reasons
+    // are: (a) Shutdown() has been called on the socket (=Stop() or Pause());
+    // (b) socket failed. In both cases go back and wait for stop or resume
+    // action.
+    callback_ = nullptr;
   }
 
-  Run();
-
   // 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));
+    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_);
       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.