Make AudioOutputDevice restartable and reinitializable

media/audio/audio_device_thread.cc

Index: media/audio/audio_device_thread.cc
diff --git a/media/audio/audio_device_thread.cc b/media/audio/audio_device_thread.cc
index f4e00d4af22b0d71dd7e6bbd82fb26ebba20b323..9e69cb2fd30efa65691b1573399242691f2ad8d6 100644
--- a/media/audio/audio_device_thread.cc
+++ b/media/audio/audio_device_thread.cc
@@ -16,6 +16,7 @@
 #include "base/memory/aligned_memory.h"
 #include "base/message_loop/message_loop.h"
 #include "base/numerics/safe_conversions.h"
+#include "base/synchronization/waitable_event.h"
 #include "base/threading/platform_thread.h"
 #include "base/threading/thread_restrictions.h"
 #include "media/base/audio_bus.h"
@@ -32,21 +33,30 @@ 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(). StartRun() 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.
   void Stop(base::MessageLoop* loop_for_join);
 
+  // Releases the thread from waiting and starts running it with the given
+  // |callback| and |socket|.
+  void StartRun(AudioDeviceThread::Callback* callback,
+                base::SyncSocket::Handle socket);
+
+  // Stops running the thread, it will wait in ThreadMain(). Shuts down the
+  // socket. StartRun() can then be called to restart it.
+  void StopRun();
+
  private:
   friend class base::RefCountedThreadSafe<AudioDeviceThread::Thread>;
+
   ~Thread() override;
 
   // Overrides from PlatformThread::Delegate.
@@ -55,13 +65,32 @@ class AudioDeviceThread::Thread
   // Runs the loop that reads from the socket.
   void Run();
 
- private:
+  // Waits for |event_| to be signaled if in IDLE state. Returns true if it did
+  // wait, otherwise false.
+  // Called on the audio thread.
+  bool MaybeWait();
+
+  // Initializes |socket_| and tells |callback_| to initialize itself.
+  // Called on the audio thread.
+  void Initialize();
+
+  enum State {
+    IDLE,
+    RUNNING,
+    STOPPING
+  };
+
   base::PlatformThreadHandle thread_;
   AudioDeviceThread::Callback* callback_;
-  base::CancelableSyncSocket socket_;
-  base::Lock callback_lock_;
+  scoped_ptr<base::CancelableSyncSocket> socket_;
+  base::SyncSocket::Handle new_socket_handle_;
+  State state_;
+  base::WaitableEvent event_;

[o1ka, 2016/02/17 09:46:29]

Can we name it more descriptive, like 'wait_for_command', or 'start_or_stop', or
something like that? It would make the code more verbal.

   const char* thread_name_;
   const bool synchronized_buffers_;
+  // TODO: Rename.
+  base::Lock callback_lock_;  // Protects |callback_| and |new_socket_|.
+  base::Lock state_lock_;  // Protects |state_|.
 
   DISALLOW_COPY_AND_ASSIGN(Thread);
 };
@@ -73,14 +102,13 @@ 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) {
+  printf("*********** Starting audio device thread\n");
   base::AutoLock auto_lock(thread_lock_);
   CHECK(!thread_.get());
   thread_ = new AudioDeviceThread::Thread(
-      callback, socket, thread_name, synchronized_buffers);
+      thread_name, synchronized_buffers);
   thread_->Start();
 }
 
@@ -92,19 +120,34 @@ void AudioDeviceThread::Stop(base::MessageLoop* loop_for_join) {
   }
 }
 
+void AudioDeviceThread::StartRun(AudioDeviceThread::Callback* callback,
+                                     base::SyncSocket::Handle socket) {
+  base::AutoLock auto_lock(thread_lock_);

[tommi (sloooow) - chröme, 2016/02/16 11:25:29]

This might be off topic but...

can we make it so that Start* and Stop* methods are called on the same thread?
I.e. that thread_ is always accessed on the same thread and therefore we don't
need a lock for it?

As is, we're holding a lock while we wait for a thread to terminate and I'm
wondering if we still need it.
I seem to be the one to blame for this btw, and back in 2012 I wrote this:

"Additionally, I'm adding a lock to guard the thread_ variable in
AudioDeviceThread.
Before I was relying on external synchronization in the AudioDevice and
AudioInputDevice state machine.  However, it doesn't hurt to go belt and
suspenders although it might not look cool ;)"

[Henrik Grunell, 2016/02/16 12:47:19]

:) Yeah, I can add a thread checker instead, let's see if it blows up.

+  CHECK(thread_.get());

[tommi (sloooow) - chröme, 2016/02/16 11:25:29]

nit: no need for these sort of checks since we deref thread_ anyway.

[Henrik Grunell, 2016/02/16 12:47:19]

Ineed. Done.

+  thread_->StartRun(callback, socket);
+}
+
+void AudioDeviceThread::StopRun() {
+  base::AutoLock auto_lock(thread_lock_);
+  // TODO: Allow calling if not started?
+//  CHECK(thread_.get());
+  if (thread_.get())
+    thread_->StopRun();
+}
+
 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),
+      callback_(nullptr),
+      new_socket_handle_(base::CancelableSyncSocket::kInvalidHandle),
+      state_(IDLE),
+      event_(false, false),
       thread_name_(thread_name),
       synchronized_buffers_(synchronized_buffers) {
 }
@@ -114,8 +157,11 @@ AudioDeviceThread::Thread::~Thread() {
 }
 
 void AudioDeviceThread::Thread::Start() {
+  printf("____________ Thread Start. thread = %0lx\n",
+         reinterpret_cast<size_t*>(&thread_)[0]);
   base::AutoLock auto_lock(callback_lock_);
   DCHECK(thread_.is_null());
+  DCHECK_EQ(state_, IDLE);
   // This reference will be released when the thread exists.
   AddRef();
 
@@ -125,7 +171,16 @@ void AudioDeviceThread::Thread::Start() {
 }
 
 void AudioDeviceThread::Thread::Stop(base::MessageLoop* loop_for_join) {
-  socket_.Shutdown();
+  printf("____________ Thread Stop. thread = %0lx\n",
+         reinterpret_cast<size_t*>(&thread_)[0]);
+  {
+    base::AutoLock auto_lock(state_lock_);
+    DCHECK(state_ == IDLE || state_ == RUNNING);

[tommi (sloooow) - chröme, 2016/02/16 11:25:29]

Can/do we update the state_ always on the same thread?

[Henrik Grunell, 2016/02/16 12:47:19]

Yes, if we add a thread checker and are sure it's called on the same thread (I
think it should be). Then it's updated only on this thread. And only read on the
audio thread.

+    state_ = STOPPING;
+  }
+
+  socket_->Shutdown();

[o1ka, 2016/02/17 09:46:29]

Is it safe to shutdown a socket twice?

+  event_.Signal();
 
   base::PlatformThreadHandle thread = base::PlatformThreadHandle();
 
@@ -145,6 +200,44 @@ void AudioDeviceThread::Thread::Stop(base::MessageLoop* loop_for_join) {
   }
 }
 
+void AudioDeviceThread::Thread::StartRun(
+    AudioDeviceThread::Callback* callback,
+    base::SyncSocket::Handle socket) {
+  printf("____________ Thread StartRun. thread = %0lx\n",
+         reinterpret_cast<size_t*>(&thread_)[0]);
+  // TODO: Could maybe use one lock for all three variables.
+  {
+    base::AutoLock auto_lock(callback_lock_);
+    callback_ = callback;
+    new_socket_handle_ = socket;
+  }
+  {
+    base::AutoLock auto_lock(state_lock_);
+    DCHECK_EQ(state_, IDLE);
+    state_ = RUNNING;
+  }
+  event_.Signal();
+}
+
+void AudioDeviceThread::Thread::StopRun() {
+  printf("____________ Thread StopRun. thread = %0lx\n",
+         reinterpret_cast<size_t*>(&thread_)[0]);
+  {
+    base::AutoLock auto_lock(state_lock_);
+    // TODO: Remove if we don't allow StopRun when not running. Should likely
+    // require a IsRunning() function, since AOD is pretty messed up.
+    if (state_ == IDLE)
+      return;
+    DCHECK_EQ(state_, RUNNING);
+    state_ = IDLE;
+  }
+  {
+    base::AutoLock auto_lock(callback_lock_);
+    callback_ = nullptr;
+  }
+  socket_->Shutdown();
+}
+
 void AudioDeviceThread::Thread::ThreadMain() {
   PlatformThread::SetName(thread_name_);
 
@@ -154,13 +247,37 @@ void AudioDeviceThread::Thread::ThreadMain() {
   // 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();
+  MaybeWait();
+  // TODO: Handle case when Stop() is called before we get here.

[o1ka, 2016/02/17 09:46:29]

For this you can probably just place if(MaybeWait()) before Run() in the loop

+  Initialize();
+
+  static_assert(sizeof(size_t) == sizeof(pthread_t), "This won't work");

[tommi (sloooow) - chröme, 2016/02/16 11:25:29]

fyi - pthread_t doesn't exist on some platforms and it doesn't always reflect
the same thing. platform_thread.h should help. check out the difference between
PlatformThreadId and PlatformThreadRef.

[Henrik Grunell, 2016/02/16 12:47:19]

Yeah, this was just for verifying my assumption in the debug printouts. Removed.

+  printf("____________ Thread Run start. thread = %0lx\n",
+         reinterpret_cast<size_t*>(&thread_)[0]);
+
+  while (true) {
+    Run();
+
+    // If StopRun() was called, we'll wait until signalled by either StartRun()
+    // or Stop().
+    if (MaybeWait()) {
+      {
+        // We must check if we are stopping, it may have been called while
+        // waiting.
+        base::AutoLock auto_lock(state_lock_);
+        if (state_ == STOPPING)
+          break;
+      }
+      Initialize();
+    } else {
+      // This happens if Stop() was called while running or if the socket
+      // operation failed.
+      break;
+    }
   }
 
-  Run();
+  printf("____________ Thread Run end. thread = %0lx\n",
+         reinterpret_cast<size_t*>(&thread_)[0]);
 
   // Release the reference for the thread. Note that after this, the Thread
   // instance will most likely be deleted.
@@ -169,11 +286,18 @@ void AudioDeviceThread::Thread::ThreadMain() {
 
 void AudioDeviceThread::Thread::Run() {
   uint32_t buffer_index = 0;
+
   while (true) {
+//    printf("____________ Thread Run. thread = %0lx, socket handle = %d\n",
+//           reinterpret_cast<size_t*>(&thread_)[0], socket_->handle());
     uint32_t pending_data = 0;
-    size_t bytes_read = socket_.Receive(&pending_data, sizeof(pending_data));
-    if (bytes_read != sizeof(pending_data))
+    size_t bytes_read = socket_->Receive(&pending_data, sizeof(pending_data));
+
+    if (bytes_read != sizeof(pending_data)) {
+      printf("____________ Thread Run receive failed. thread = %0lx, socket handle = %d\n",
+             reinterpret_cast<size_t*>(&thread_)[0], socket_->handle());
       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
@@ -186,6 +310,8 @@ void AudioDeviceThread::Thread::Run() {
     if (pending_data != std::numeric_limits<uint32_t>::max()) {
       base::AutoLock auto_lock(callback_lock_);
       if (callback_)
+//        printf("____________ Thread Run callback. thread = %0lx, socket handle = %d\n",
+//               reinterpret_cast<size_t*>(&thread_)[0], socket_->handle());
         callback_->Process(pending_data);
     }
 
@@ -201,13 +327,67 @@ void AudioDeviceThread::Thread::Run() {
     // details on how this works see AudioSyncReader::WaitUntilDataIsReady().
     if (synchronized_buffers_) {
       ++buffer_index;
-      size_t bytes_sent = socket_.Send(&buffer_index, sizeof(buffer_index));
-      if (bytes_sent != sizeof(buffer_index))
+      size_t bytes_sent = socket_->Send(&buffer_index, sizeof(buffer_index));
+
+      if (bytes_sent != sizeof(buffer_index)) {
+        printf("____________ Thread Run send failed. thread = %0lx, socket handle = %d\n",
+               reinterpret_cast<size_t*>(&thread_)[0], socket_->handle());
         break;
+      }
     }
   }
 }
 
+bool AudioDeviceThread::Thread::MaybeWait() {

[o1ka, 2016/02/17 09:46:29]

I would call it "Continue" or "Resume" or "SyncAndResume" or something like this
to explain its return value meaning.

+  int socket_handle = socket_ ? socket_->handle()
+                              : base::CancelableSyncSocket::kInvalidHandle;
+  printf("____________ Thread Maybe wait. thread = %0lx, socket handle = %d\n",
+         reinterpret_cast<size_t*>(&thread_)[0], socket_handle);
+
+  State state;
+  {
+    base::AutoLock auto_lock(state_lock_);

[tommi (sloooow) - chröme, 2016/02/16 11:25:29]

this can be a wait too

[Henrik Grunell, 2016/02/16 12:47:19]

Wait for the lock you mean?

[o1ka, 2016/02/17 09:46:29]

If state modifications are done by this thread only, than lock is not needed,
otherwise looks like a race is possible between setting a state and making a
decision on 'to wait or not to wait'. 

+    state = state_;
+  }
+
+  if (state == IDLE) {
+    printf("____________ Thread waiting. thread = %0lx, socket handle = %d\n",
+           reinterpret_cast<size_t*>(&thread_)[0], socket_handle);
+    event_.Wait();
+    printf("____________ Thread waiting done. thread = %0lx, socket handle = %d\n",
+           reinterpret_cast<size_t*>(&thread_)[0], socket_handle);
+    return true;
+  }
+
+  // Make sure the event is reset when returning. This is necessary when for
+  // example StartRun is called before ThreadMain() has started, in which case
+  // |state_| == RUNNING and the event is signaled.

[tommi (sloooow) - chröme, 2016/02/16 11:25:29]

This reads like a bug to me actually.  The event should be auto resetting and
the race described in the TODO exists only because we're trying to reset
manually.

What's the difference between Start and StartRun?

[Henrik Grunell, 2016/02/16 12:47:19]

Yes, that's what I want to do, but the case when the socket fails without us
stopping may be problematic.
Start creates the thread, but it will wait here above until StartRun is called.
We could make Start not wait too, I'm leaning against that but waited for input
first. :)

+  // TODO: What if context switch right before the below line and then event
+  // is signaled on the other thread. That needs to be fixed. Can maybe the
+  // state lock be held when waiting?
+  // Ideally remove |state_| and only use the event. The problem then is that
+  // we'll wait on the event if the socket receive or sen fails. At least don't
+  // reset on this thread.
+  event_.Reset();
+  return false;
+}
+
+void AudioDeviceThread::Thread::Initialize() {
+  int socket_handle = socket_ ? socket_->handle()
+                              : base::CancelableSyncSocket::kInvalidHandle;
+  printf("____________ Thread init. thread = %0lx, socket handle = %d\n",
+         reinterpret_cast<size_t*>(&thread_)[0], socket_handle);
+  base::AutoLock auto_lock(callback_lock_);

[tommi (sloooow) - chröme, 2016/02/16 11:25:29]

any chance we can get away with not using this lock? (socket_ always touched on
the same thread)

[Henrik Grunell, 2016/02/16 12:47:19]

|callback_| needs protection. (And |new_socket_handle_|.)

+  DCHECK_NE(new_socket_handle_, base::CancelableSyncSocket::kInvalidHandle);
+  socket_.reset(
+      new base::CancelableSyncSocket(new_socket_handle_));
+  new_socket_handle_ = base::CancelableSyncSocket::kInvalidHandle;
+  // TODO: We should bail out earlier (i.e. not initialize |socket_|) if no
+  // |callback_|
+  if (callback_)
+    callback_->InitializeOnAudioThread();
+}
+
 // AudioDeviceThread::Callback implementation
 
 AudioDeviceThread::Callback::Callback(