// Copyright 2015 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "chromecast/media/audio/cast_audio_output_stream.h"
#include <string>
#include <utility>
#include "base/bits.h"
#include "base/functional/bind.h"
#include "base/functional/callback.h"
#include "base/functional/callback_helpers.h"
#include "base/location.h"
#include "base/logging.h"
#include "base/message_loop/message_pump_type.h"
#include "base/synchronization/lock.h"
#include "base/synchronization/waitable_event.h"
#include "base/task/bind_post_task.h"
#include "base/task/single_thread_task_runner.h"
#include "base/time/time.h"
#include "chromecast/base/bind_to_task_runner.h"
#include "chromecast/base/metrics/cast_metrics_helper.h"
#include "chromecast/media/api/cma_backend_factory.h"
#include "chromecast/media/audio/cast_audio_manager.h"
#include "chromecast/media/audio/cast_audio_output_utils.h"
#include "chromecast/media/audio/cma_audio_output_stream.h"
#include "chromecast/media/audio/mixer_service/mixer_service_transport.pb.h"
#include "chromecast/media/audio/mixer_service/output_stream_connection.h"
#include "chromecast/media/audio/net/common.pb.h"
#include "chromecast/public/cast_media_shlib.h"
#include "chromecast/public/media/decoder_config.h"
#include "chromecast/public/media/media_pipeline_device_params.h"
#include "chromecast/public/volume_control.h"
#include "media/audio/audio_device_description.h"
#define POST_TO_CMA_WRAPPER(method, ...) \
do { \
DCHECK(cma_wrapper_); \
audio_manager_->media_task_runner()->PostTask( \
FROM_HERE, \
base::BindOnce(&CmaAudioOutputStream::method, \
base::Unretained(cma_wrapper_.get()), ##__VA_ARGS__)); \
} while (0)
#define POST_TO_MIXER_SERVICE_WRAPPER(method, ...) \
do { \
DCHECK(mixer_service_wrapper_); \
mixer_service_wrapper_->io_task_runner()->PostTask( \
FROM_HERE, \
base::BindOnce(&MixerServiceWrapper::method, \
base::Unretained(mixer_service_wrapper_.get()), \
##__VA_ARGS__)); \
} while (0)
namespace {
// Below are settings for MixerService and the DirectAudio it uses.
constexpr base::TimeDelta kFadeTime = base::Milliseconds(5);
constexpr base::TimeDelta kCommunicationsMaxBufferedFrames =
base::Milliseconds(50);
constexpr base::TimeDelta kMediaMaxBufferedFrames = base::Milliseconds(70);
} // namespace
namespace chromecast {
namespace media {
namespace {
AudioContentType GetContentType(const std::string& device_id) {
if (::media::AudioDeviceDescription::IsCommunicationsDevice(device_id)) {
return AudioContentType::kCommunication;
}
return AudioContentType::kMedia;
}
audio_service::ContentType ConvertContentType(AudioContentType content_type) {
switch (content_type) {
case AudioContentType::kMedia:
return audio_service::CONTENT_TYPE_MEDIA;
case AudioContentType::kCommunication:
return audio_service::CONTENT_TYPE_COMMUNICATION;
default:
NOTREACHED_IN_MIGRATION();
return audio_service::CONTENT_TYPE_MEDIA;
}
}
} // namespace
class CastAudioOutputStream::MixerServiceWrapper
: public mixer_service::OutputStreamConnection::Delegate {
public:
MixerServiceWrapper(const ::media::AudioParameters& audio_params,
const std::string& device_id);
MixerServiceWrapper(const MixerServiceWrapper&) = delete;
MixerServiceWrapper& operator=(const MixerServiceWrapper&) = delete;
~MixerServiceWrapper() override = default;
void SetRunning(bool running);
void Start(AudioSourceCallback* source_callback);
void Stop(base::WaitableEvent* finished);
void Close(base::OnceClosure closure);
void SetVolume(double volume);
int64_t GetMaxBufferedFrames();
void Flush();
base::SingleThreadTaskRunner* io_task_runner() {
return io_task_runner_.get();
}
private:
// mixer_service::OutputStreamConnection::Delegate implementation:
void FillNextBuffer(void* buffer,
int frames,
int64_t delay_timestamp,
int64_t delay) override;
// We don't push an EOS buffer.
void OnEosPlayed() override { NOTREACHED_IN_MIGRATION(); }
const ::media::AudioParameters audio_params_;
const std::string device_id_;
std::unique_ptr<::media::AudioBus> audio_bus_;
AudioSourceCallback* source_callback_;
std::unique_ptr<mixer_service::OutputStreamConnection> mixer_connection_;
double volume_;
int64_t max_buffered_frames_;
base::Lock running_lock_;
bool running_ = true;
// MixerServiceWrapper must run on an "io thread".
base::Thread io_thread_;
// Task runner on |io_thread_|.
scoped_refptr<base::SingleThreadTaskRunner> io_task_runner_;
THREAD_CHECKER(io_thread_checker_);
};
CastAudioOutputStream::MixerServiceWrapper::MixerServiceWrapper(
const ::media::AudioParameters& audio_params,
const std::string& device_id)
: audio_params_(audio_params),
device_id_(device_id),
source_callback_(nullptr),
volume_(1.0f),
max_buffered_frames_(GetMaxBufferedFrames()),
io_thread_("CastAudioOutputStream IO") {
DETACH_FROM_THREAD(io_thread_checker_);
base::Thread::Options options;
options.message_pump_type = base::MessagePumpType::IO;
options.thread_type = base::ThreadType::kRealtimeAudio;
CHECK(io_thread_.StartWithOptions(std::move(options)));
io_task_runner_ = io_thread_.task_runner();
DCHECK(io_task_runner_);
}
void CastAudioOutputStream::MixerServiceWrapper::SetRunning(bool running) {
base::AutoLock lock(running_lock_);
running_ = running;
}
void CastAudioOutputStream::MixerServiceWrapper::Start(
AudioSourceCallback* source_callback) {
DCHECK_CALLED_ON_VALID_THREAD(io_thread_checker_);
mixer_service::OutputStreamParams params;
params.set_content_type(audio_service::CONTENT_TYPE_MEDIA);
params.set_focus_type(ConvertContentType(GetContentType(device_id_)));
params.set_device_id(device_id_);
params.set_stream_type(
mixer_service::OutputStreamParams::STREAM_TYPE_DEFAULT);
params.set_sample_format(audio_service::SAMPLE_FORMAT_FLOAT_P);
params.set_sample_rate(audio_params_.sample_rate());
params.set_num_channels(audio_params_.channels());
params.set_start_threshold_frames(max_buffered_frames_);
params.set_max_buffered_frames(max_buffered_frames_);
params.set_fill_size_frames(audio_params_.frames_per_buffer());
params.set_fade_frames(::media::AudioTimestampHelper::TimeToFrames(
kFadeTime, audio_params_.sample_rate()));
params.set_use_start_timestamp(false);
source_callback_ = source_callback;
mixer_connection_ =
std::make_unique<mixer_service::OutputStreamConnection>(this, params);
mixer_connection_->Connect();
mixer_connection_->SetVolumeMultiplier(volume_);
}
void CastAudioOutputStream::MixerServiceWrapper::Stop(
base::WaitableEvent* finished) {
DCHECK_CALLED_ON_VALID_THREAD(io_thread_checker_);
mixer_connection_.reset();
source_callback_ = nullptr;
if (finished) {
finished->Signal();
}
}
void CastAudioOutputStream::MixerServiceWrapper::Flush() {
DCHECK_CALLED_ON_VALID_THREAD(io_thread_checker_);
// Nothing to do.
return;
}
void CastAudioOutputStream::MixerServiceWrapper::Close(
base::OnceClosure closure) {
DCHECK_CALLED_ON_VALID_THREAD(io_thread_checker_);
Stop(nullptr);
std::move(closure).Run();
}
int64_t CastAudioOutputStream::MixerServiceWrapper::GetMaxBufferedFrames() {
int fill_size_frames = audio_params_.frames_per_buffer();
base::TimeDelta target_max_buffered_ms = kMediaMaxBufferedFrames;
if (GetContentType(device_id_) == AudioContentType::kCommunication) {
target_max_buffered_ms = kCommunicationsMaxBufferedFrames;
}
int64_t target_max_buffered_frames =
::media::AudioTimestampHelper::TimeToFrames(target_max_buffered_ms,
audio_params_.sample_rate());
// Calculate the buffer size necessary to achieve at least the desired buffer
// duration, while minimizing latency.
int64_t max_buffered_frames = 0;
if (fill_size_frames > target_max_buffered_frames) {
max_buffered_frames = target_max_buffered_frames;
} else {
// Find the largest multiple of |fill_size_frames| that is still no larger
// than |target_max_buffered_frames|.
max_buffered_frames =
(target_max_buffered_frames / fill_size_frames) * fill_size_frames;
}
if (max_buffered_frames != target_max_buffered_frames) {
max_buffered_frames += 1;
}
return max_buffered_frames;
}
void CastAudioOutputStream::MixerServiceWrapper::SetVolume(double volume) {
DCHECK_CALLED_ON_VALID_THREAD(io_thread_checker_);
volume_ = volume;
if (mixer_connection_)
mixer_connection_->SetVolumeMultiplier(volume_);
}
void CastAudioOutputStream::MixerServiceWrapper::FillNextBuffer(
void* buffer,
int frames,
int64_t delay_timestamp,
int64_t delay) {
DCHECK_CALLED_ON_VALID_THREAD(io_thread_checker_);
// Round down to closest multiple of 4 to ensure correct channel alignment.
frames = base::bits::AlignDownDeprecatedDoNotUse(frames, 4);
// Acquire running_lock_ for the scope of this fill call to
// prevent the source callback from closing the output stream
// mid-fill.
base::AutoLock lock(running_lock_);
// Do not fill more buffers if we have stopped running.
if (!running_)
return;
int64_t playout_timestamp =
(delay_timestamp == INT64_MIN ? INT64_MIN : delay_timestamp + delay);
if (playout_timestamp < 0) {
// Assume any negative timestamp is invalid.
playout_timestamp = 0;
}
// Wrap the data buffer so we can write directly into it.
if (!audio_bus_) {
audio_bus_ = ::media::AudioBus::CreateWrapper(audio_params_.channels());
}
float* channel_data = static_cast<float*>(buffer);
for (int c = 0; c < audio_params_.channels(); ++c) {
audio_bus_->SetChannelData(c, channel_data + c * frames);
}
audio_bus_->set_frames(frames);
base::TimeDelta reported_delay = ::media::AudioTimestampHelper::FramesToTime(
max_buffered_frames_, audio_params_.sample_rate());
base::TimeTicks reported_delay_timestamp =
base::TimeTicks() + base::Microseconds(playout_timestamp);
int frames_filled = source_callback_->OnMoreData(
reported_delay, reported_delay_timestamp, {}, audio_bus_.get());
DCHECK_EQ(frames_filled, frames);
mixer_connection_->SendNextBuffer(frames);
}
CastAudioOutputStream::CastAudioOutputStream(
CastAudioManagerHelper* audio_manager,
const ::media::AudioParameters& audio_params,
const std::string& device_id_or_group_id,
bool use_mixer_service)
: volume_(1.0),
audio_thread_state_(AudioOutputState::kClosed),
audio_manager_(audio_manager),
audio_params_(audio_params),
device_id_(IsValidDeviceId(device_id_or_group_id)
? device_id_or_group_id
: ::media::AudioDeviceDescription::kDefaultDeviceId),
group_id_(GetGroupId(device_id_or_group_id)),
use_mixer_service_(use_mixer_service),
audio_weak_factory_(this) {
DCHECK(audio_manager_);
DETACH_FROM_THREAD(audio_thread_checker_);
DVLOG(1) << __func__ << " " << this << " created from group_id=" << group_id_
<< " with audio_params=" << audio_params_.AsHumanReadableString();
audio_weak_this_ = audio_weak_factory_.GetWeakPtr();
}
CastAudioOutputStream::~CastAudioOutputStream() {
DCHECK_CALLED_ON_VALID_THREAD(audio_thread_checker_);
}
bool CastAudioOutputStream::Open() {
DCHECK_CALLED_ON_VALID_THREAD(audio_thread_checker_);
DVLOG(1) << this << ": " << __func__;
if (audio_thread_state_ != AudioOutputState::kClosed)
return false;
// Sanity check the audio parameters.
::media::AudioParameters::Format format = audio_params_.format();
DCHECK((format == ::media::AudioParameters::AUDIO_PCM_LINEAR) ||
(format == ::media::AudioParameters::AUDIO_PCM_LOW_LATENCY));
::media::ChannelLayout channel_layout = audio_params_.channel_layout();
if ((channel_layout != ::media::CHANNEL_LAYOUT_MONO) &&
(channel_layout != ::media::CHANNEL_LAYOUT_STEREO)) {
LOG(WARNING) << "Unsupported channel layout: " << channel_layout;
return false;
}
DCHECK_GE(audio_params_.channels(), 1);
DCHECK_LE(audio_params_.channels(), 2);
const std::string application_session_id =
audio_manager_->GetSessionId(group_id_);
LOG_IF(WARNING, application_session_id.empty()) << "Session id is empty.";
DVLOG(1) << this << ": " << __func__
<< ", session_id=" << application_session_id;
if (!use_mixer_service_) {
cma_wrapper_ = std::make_unique<CmaAudioOutputStream>(
audio_params_, audio_params_.GetBufferDuration(), device_id_,
audio_manager_->GetCmaBackendFactory());
POST_TO_CMA_WRAPPER(Initialize, application_session_id);
} else {
DCHECK(!(audio_params_.effects() & ::media::AudioParameters::MULTIZONE));
mixer_service_wrapper_ =
std::make_unique<MixerServiceWrapper>(audio_params_, device_id_);
}
audio_thread_state_ = AudioOutputState::kOpened;
return true;
}
void CastAudioOutputStream::Close() {
DCHECK_CALLED_ON_VALID_THREAD(audio_thread_checker_);
DVLOG(1) << this << ": " << __func__;
audio_thread_state_ = AudioOutputState::kPendingClose;
base::OnceClosure finish_callback = BindToCurrentThread(
base::BindOnce(&CastAudioOutputStream::FinishClose, audio_weak_this_));
if (mixer_service_wrapper_) {
// Synchronously set running to false to guarantee that
// AudioSourceCallback::OnMoreData() will not be called anymore.
mixer_service_wrapper_->SetRunning(false);
POST_TO_MIXER_SERVICE_WRAPPER(
Close,
base::BindPostTask(audio_manager_->audio_manager()->GetTaskRunner(),
std::move(finish_callback)));
} else if (cma_wrapper_) {
// Synchronously set running to false to guarantee that
// AudioSourceCallback::OnMoreData() will not be called anymore.
cma_wrapper_->SetRunning(false);
POST_TO_CMA_WRAPPER(Close, std::move(finish_callback));
} else {
std::move(finish_callback).Run();
}
}
void CastAudioOutputStream::FinishClose() {
DCHECK_CALLED_ON_VALID_THREAD(audio_thread_checker_);
// Signal to the manager that we're closed and can be removed.
// This should be the last call during the close process as it deletes "this".
audio_manager_->audio_manager()->ReleaseOutputStream(this);
}
void CastAudioOutputStream::Start(AudioSourceCallback* source_callback) {
DCHECK(source_callback);
DCHECK_CALLED_ON_VALID_THREAD(audio_thread_checker_);
// We allow calls to start even in the unopened state.
DCHECK_NE(audio_thread_state_, AudioOutputState::kPendingClose);
DVLOG(2) << this << ": " << __func__;
audio_thread_state_ = AudioOutputState::kStarted;
metrics::CastMetricsHelper::GetInstance()->LogTimeToFirstAudio();
// |cma_wrapper_| and |mixer_service_wrapper_| cannot be both active.
DCHECK(!(cma_wrapper_ && mixer_service_wrapper_));
if (cma_wrapper_) {
POST_TO_CMA_WRAPPER(Start, source_callback);
} else {
POST_TO_MIXER_SERVICE_WRAPPER(Start, source_callback);
}
}
void CastAudioOutputStream::Stop() {
DCHECK_CALLED_ON_VALID_THREAD(audio_thread_checker_);
DVLOG(2) << this << ": " << __func__;
// We allow calls to stop even in the unstarted/unopened state.
if (audio_thread_state_ != AudioOutputState::kStarted)
return;
audio_thread_state_ = AudioOutputState::kOpened;
// |cma_wrapper_| and |mixer_service_wrapper_| cannot be both active.
DCHECK(!(cma_wrapper_ && mixer_service_wrapper_));
base::WaitableEvent finished;
if (cma_wrapper_) {
POST_TO_CMA_WRAPPER(Stop, &finished);
} else if (mixer_service_wrapper_) {
POST_TO_MIXER_SERVICE_WRAPPER(Stop, &finished);
} else {
finished.Signal();
}
finished.Wait();
}
void CastAudioOutputStream::Flush() {
DCHECK_CALLED_ON_VALID_THREAD(audio_thread_checker_);
DVLOG(2) << this << ": " << __func__;
// |cma_wrapper_| and |mixer_service_wrapper_| cannot be both active.
DCHECK(!(cma_wrapper_ && mixer_service_wrapper_));
if (cma_wrapper_) {
// Make sure this is not on the same thread as CMA_WRAPPER to prevent
// deadlock.
DCHECK(!audio_manager_->media_task_runner()->BelongsToCurrentThread());
base::WaitableEvent finished;
POST_TO_CMA_WRAPPER(Flush, base::Unretained(&finished));
finished.Wait();
} else if (mixer_service_wrapper_) {
POST_TO_MIXER_SERVICE_WRAPPER(Flush);
}
}
void CastAudioOutputStream::SetVolume(double volume) {
DCHECK_CALLED_ON_VALID_THREAD(audio_thread_checker_);
DCHECK_NE(audio_thread_state_, AudioOutputState::kPendingClose);
DVLOG(2) << this << ": " << __func__ << "(" << volume << ")";
volume_ = volume;
DCHECK(!(cma_wrapper_ && mixer_service_wrapper_));
if (cma_wrapper_) {
POST_TO_CMA_WRAPPER(SetVolume, volume);
} else {
POST_TO_MIXER_SERVICE_WRAPPER(SetVolume, volume);
}
}
void CastAudioOutputStream::GetVolume(double* volume) {
DCHECK_CALLED_ON_VALID_THREAD(audio_thread_checker_);
*volume = volume_;
}
} // namespace media
} // namespace chromecast
Codebase Browser
chromium