// 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 <algorithm>
#include "base/bind.h"
#include "base/logging.h"
#include "base/memory/aligned_memory.h"
#include "base/message_loop/message_loop.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);
void Start();
// Stops 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);
private:
friend class base::RefCountedThreadSafe<AudioDeviceThread::Thread>;
virtual ~Thread();
// Overrides from PlatformThread::Delegate.
virtual void ThreadMain() OVERRIDE;
// Runs the loop that reads from the socket.
void Run();
private:
base::PlatformThreadHandle thread_;
AudioDeviceThread::Callback* callback_;
base::CancelableSyncSocket socket_;
base::Lock callback_lock_;
const char* thread_name_;
const bool synchronized_buffers_;
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) {
base::AutoLock auto_lock(thread_lock_);
CHECK(!thread_);
thread_ = new AudioDeviceThread::Thread(
callback, socket, 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;
}
}
bool AudioDeviceThread::IsStopped() {
base::AutoLock auto_lock(thread_lock_);
return !thread_;
}
// AudioDeviceThread::Thread implementation
AudioDeviceThread::Thread::Thread(AudioDeviceThread::Callback* callback,
base::SyncSocket::Handle socket,
const char* thread_name,
bool synchronized_buffers)
: thread_(),
callback_(callback),
socket_(socket),
thread_name_(thread_name),
synchronized_buffers_(synchronized_buffers) {
}
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.
AddRef();
PlatformThread::CreateWithPriority(0, this, &thread_,
base::kThreadPriority_RealtimeAudio);
CHECK(!thread_.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()) {
if (loop_for_join) {
loop_for_join->PostTask(FROM_HERE,
base::Bind(&base::PlatformThread::Join, thread));
} else {
base::PlatformThread::Join(thread);
}
}
}
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();
}
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 buffer_index = 0;
while (true) {
int pending_data = 0;
size_t bytes_read = socket_.Receive(&pending_data, sizeof(pending_data));
if (bytes_read != sizeof(pending_data)) {
DCHECK_EQ(bytes_read, 0U);
break;
}
{
base::AutoLock auto_lock(callback_lock_);
if (callback_)
callback_->Process(pending_data);
}
// 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))
break;
}
}
}
// 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() *
(audio_parameters_.bits_per_sample() / 8) *
samples_per_ms_),
shared_memory_(memory, false),
memory_length_(memory_length),
total_segments_(total_segments) {
CHECK_NE(bytes_per_ms_, 0); // Catch division by zero early.
CHECK_NE(samples_per_ms_, 0);
CHECK_GT(total_segments_, 0);
CHECK_EQ(memory_length_ % total_segments_, 0);
segment_length_ = memory_length_ / total_segments_;
}
AudioDeviceThread::Callback::~Callback() {}
void AudioDeviceThread::Callback::InitializeOnAudioThread() {
MapSharedMemory();
CHECK(shared_memory_.memory());
}
} // namespace media.