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/waitable_event.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(). 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.
   void ThreadMain() override;
 
   // 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);
 };
 
 // AudioDeviceThread implementation
 
 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();
 }
 
 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::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) {
 }
 
 AudioDeviceThread::Thread::~Thread() {
   DCHECK(thread_.is_null());
 }
 
 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();
 
   PlatformThread::CreateWithPriority(0, this, &thread_,
                                      base::ThreadPriority::REALTIME_AUDIO);
   CHECK(!thread_.is_null());
 }
 
 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();
 
   {  // NOLINT
     base::AutoLock auto_lock(callback_lock_);
     callback_ = NULL;
     std::swap(thread, thread_);
   }
 
   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::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_);
 
   // 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();
+  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.
   Release();
 }
 
 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
     // 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_)
+//        printf("____________ Thread Run callback. thread = %0lx, socket handle = %d\n",
+//               reinterpret_cast<size_t*>(&thread_)[0], socket_->handle());
         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));
-      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(
     const AudioParameters& audio_parameters,
     base::SharedMemoryHandle memory,
     int memory_length,
     int total_segments)
     : audio_parameters_(audio_parameters),
       samples_per_ms_(audio_parameters.sample_rate() / 1000),
       bytes_per_ms_(audio_parameters.channels() *
@@ skipping 10 matching lines @@
 }
 
 AudioDeviceThread::Callback::~Callback() {}
 
 void AudioDeviceThread::Callback::InitializeOnAudioThread() {
   MapSharedMemory();
   CHECK(shared_memory_.memory());
 }
 
 }  // namespace media.