dmaax/ladybird
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

LibMedia: Transition producers and sinks to the new pipeline model

Browse source 
This is an intermediate step towards unifying the pipeline around new
Producer/Sink interfaces. Producers now have a pull() method that gets
the next piece of data from them. The pull() method returns a status
that can indicate whether it has current data, and if not, why it's
unavailable. This signal will be passed down the pipeline to the final
sink, which can expose the signal to its user, which in the normal
playback pipeline is PlaybackManager. The signal can be used to
transition between playback states. Currently, this is only hooked up
to the buffering state, but should be used later for ending playback
as well as decoding error propagation.

Buffering is now determined solely based on whether the pipeline is
blocked on incomplete data, so the ready state for video now progresses
past HAVE_METADATA immediately after playback manager initializes. This
will change when files have buffered ranges.
This commit is contained in:
Zaggy1024 2026-04-26 05:49:40 -05:00 committed by Gregory Bertilson
parent a60db31902
commit 7be0ae89e4
21 changed files with 483 additions and 400 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
Libraries/LibMedia/AudioBlock.h
View file

@ -6,10 +6,10 @@
#pragma once
#include <AK/FixedArray.h>
#include <AK/Math.h>
#include <AK/SaturatingMath.h>
#include <AK/Time.h>
#include <AK/Vector.h>
#include <LibMedia/Audio/SampleSpecification.h>
namespace Media {
@ -30,7 +30,7 @@ public:
{
m_sample_specification = {};
m_timestamp_in_samples = 0;
m_data = Data();
m_data.clear_with_capacity();
}
template<typename Callback>
void emplace(Audio::SampleSpecification sample_specification, AK::Duration timestamp, Callback data_callback)

76
Libraries/LibMedia/PipelineStatus.h Normal file
View file

@ -0,0 +1,76 @@
/*
* Copyright (c) 2026-present, the Ladybird developers.
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#pragma once
#include <AK/Format.h>
#include <AK/Function.h>
#include <AK/Types.h>
namespace Media {
enum class PipelineStatus : u8 {
Pending,
HaveData,
Blocked,
EndOfStream,
Error,
};
constexpr bool can_carry_data(PipelineStatus status)
{
if (status == PipelineStatus::HaveData)
return true;
if (status == PipelineStatus::EndOfStream)
return true;
return false;
}
constexpr PipelineStatus select_combined_pipeline_status(PipelineStatus a, PipelineStatus b)
{
if (a == PipelineStatus::Error || b == PipelineStatus::Error)
return PipelineStatus::Error;
if (a == PipelineStatus::Blocked || b == PipelineStatus::Blocked)
return PipelineStatus::Blocked;
if (a == PipelineStatus::HaveData || b == PipelineStatus::HaveData)
return PipelineStatus::HaveData;
if (a == PipelineStatus::Pending || b == PipelineStatus::Pending)
return PipelineStatus::Pending;
return PipelineStatus::EndOfStream;
}
using PipelineStateChangeHandler = Function<void(PipelineStatus)>;
constexpr StringView pipeline_status_to_string(PipelineStatus status)
{
switch (status) {
case PipelineStatus::Pending:
return "Pending"sv;
case PipelineStatus::HaveData:
return "HaveData"sv;
case PipelineStatus::Blocked:
return "Blocked"sv;
case PipelineStatus::EndOfStream:
return "EndOfStream"sv;
case PipelineStatus::Error:
return "Error"sv;
}
return "Invalid"sv;
}
}
namespace AK {
template<>
struct Formatter<Media::PipelineStatus> final : Formatter<StringView> {
ErrorOr<void> format(FormatBuilder& builder, Media::PipelineStatus state)
{
return Formatter<StringView>::format(builder, Media::pipeline_status_to_string(state));
}
};
}

84
Libraries/LibMedia/PlaybackManager.cpp
View file

@ -117,7 +117,12 @@ DecoderErrorOr<void> PlaybackManager::prepare_playback_from_demuxer(WeakPlayback
if (!self->m_audio_output_disabled && !self->m_audio_sink && !self->m_audio_tracks.is_empty()) {
self->m_audio_mixer = MUST(AudioMixer::try_create());
self->m_audio_sink = MUST(AudioPlaybackSink::try_create(*self->m_audio_mixer));
self->m_audio_sink = MUST(AudioPlaybackSink::try_create(*self->m_audio_mixer,
[self](PipelineStatus status) {
if (!self)
return;
self->on_audio_sink_state_changed(status);
}));
self->set_time_provider(*self->m_audio_sink);
self->m_audio_sink->on_audio_output_error = [self](Error&& error) {
if (!self)
@ -125,11 +130,6 @@ DecoderErrorOr<void> PlaybackManager::prepare_playback_from_demuxer(WeakPlayback
dbgln("Audio output initialization failed with error: {}", error);
self->disable_audio();
};
self->m_audio_mixer->on_start_buffering = [self](Track const& track) {
if (!self)
return;
self->track_started_buffering(track);
};
}
if (self->on_track_added) {
@ -216,14 +216,9 @@ WeakPlaybackManager PlaybackManager::weak()
void PlaybackManager::set_up_producers()
{
for (auto const& video_track_data : m_video_track_datas) {
auto track = video_track_data.track;
video_track_data.producer->set_error_handler([self = weak(), track](DecoderError&& error) {
video_track_data.producer->set_error_handler([self = weak()](DecoderError&& error) {
if (!self)
return;
if (error.category() == DecoderErrorCategory::EndOfStream) {
self->track_stopped_buffering(track);
return;
}
self->dispatch_error(move(error));
});
video_track_data.producer->set_duration_change_handler([self = weak()](AK::Duration time) {
@ -231,22 +226,12 @@ void PlaybackManager::set_up_producers()
return;
self->check_for_duration_change(time);
});
video_track_data.producer->set_frames_queue_is_full_handler([self = weak(), track] {
if (!self)
return;
self->track_stopped_buffering(track);
});
}
for (auto const& audio_track_data : m_audio_track_datas) {
auto track = audio_track_data.track;
audio_track_data.producer->set_error_handler([self = weak(), track](DecoderError&& error) {
audio_track_data.producer->set_error_handler([self = weak()](DecoderError&& error) {
if (!self)
return;
if (error.category() == DecoderErrorCategory::EndOfStream) {
self->track_stopped_buffering(track);
return;
}
self->dispatch_error(move(error));
});
audio_track_data.producer->set_duration_change_handler([self = weak()](AK::Duration time) {
@ -254,24 +239,35 @@ void PlaybackManager::set_up_producers()
return;
self->check_for_duration_change(time);
});
audio_track_data.producer->set_queue_is_full_handler([self = weak(), track] {
if (!self)
return;
self->track_stopped_buffering(track);
});
}
}
void PlaybackManager::track_started_buffering(Track const& track)
void PlaybackManager::on_audio_sink_state_changed(PipelineStatus status)
{
m_tracks_still_buffering.set(track);
m_handler->enter_buffering();
m_audio_buffering = status == PipelineStatus::Blocked;
update_buffering_state();
}
void PlaybackManager::track_stopped_buffering(Track const& track)
void PlaybackManager::on_video_sink_state_changed(Track const& track, PipelineStatus status)
{
m_tracks_still_buffering.remove(track);
if (m_tracks_still_buffering.is_empty())
if (status == PipelineStatus::Blocked) {
if (m_video_tracks_buffering.set(track) == HashSetResult::InsertedNewEntry)
update_buffering_state();
} else {
if (m_video_tracks_buffering.remove(track))
update_buffering_state();
}
}
void PlaybackManager::update_buffering_state()
{
auto is_buffering = m_audio_buffering || !m_video_tracks_buffering.is_empty();
if (is_buffering == m_was_buffering)
return;
m_was_buffering = is_buffering;
if (is_buffering)
m_handler->enter_buffering();
else
m_handler->exit_buffering();
}
@ -311,24 +307,24 @@ void PlaybackManager::set_time_provider(NonnullRefPtr<MediaTimeProvider> const&
void PlaybackManager::disable_audio()
{
m_audio_buffering = false;
m_audio_mixer = nullptr;
m_audio_sink = nullptr;
set_time_provider(make_ref_counted<GenericTimeProvider>());
for (auto const& track : m_audio_tracks)
track_stopped_buffering(track);
on_audio_sink_state_changed(PipelineStatus::EndOfStream);
}
NonnullRefPtr<DisplayingVideoSink> PlaybackManager::get_or_create_the_displaying_video_sink_for_track(Track const& track)
{
auto& track_data = get_video_data_for_track(track);
if (track_data.display == nullptr) {
track_data.display = MUST(Media::DisplayingVideoSink::try_create(m_time_provider));
track_data.display = MUST(Media::DisplayingVideoSink::try_create(m_time_provider,
[self = weak(), track](PipelineStatus status) {
if (!self)
return;
self->on_video_sink_state_changed(track, status);
}));
track_data.display->set_producer(track, track_data.producer);
track_data.display->m_on_start_buffering = [this, track] {
track_started_buffering(track);
};
m_tracks_still_buffering.set(track);
m_handler->on_track_enabled(track);
}
@ -342,8 +338,8 @@ void PlaybackManager::remove_the_displaying_video_sink_for_track(Track const& tr
VERIFY(track_data.display);
track_data.display->set_producer(track, nullptr);
track_data.display = nullptr;
track_stopped_buffering(track);
m_handler->on_track_disabled(track);
on_video_sink_state_changed(track, PipelineStatus::EndOfStream);
}
void PlaybackManager::enable_an_audio_track(Track const& track)
@ -354,7 +350,6 @@ void PlaybackManager::enable_an_audio_track(Track const& track)
if (m_audio_mixer) {
VERIFY(m_audio_mixer->producer(track) == nullptr);
m_audio_mixer->set_producer(track, track_data.producer);
m_tracks_still_buffering.set(track);
}
m_handler->on_track_enabled(track);
}
@ -368,7 +363,6 @@ void PlaybackManager::disable_an_audio_track(Track const& track)
m_audio_mixer->set_producer(track, nullptr);
}
track_data.enabled = false;
track_stopped_buffering(track);
m_handler->on_track_disabled(track);
}

10
Libraries/LibMedia/PlaybackManager.h
View file

@ -18,6 +18,7 @@
#include <LibMedia/Export.h>
#include <LibMedia/Forward.h>
#include <LibMedia/MediaTimeProvider.h>
#include <LibMedia/PipelineStatus.h>
#include <LibMedia/PlaybackStates/Forward.h>
#include <LibMedia/PlaybackStates/PlaybackState.h>
#include <LibMedia/TimeRanges.h>
@ -125,8 +126,9 @@ private:
void disable_audio();
void set_up_producers();
void track_started_buffering(Track const&);
void track_stopped_buffering(Track const&);
void on_audio_sink_state_changed(PipelineStatus);
void on_video_sink_state_changed(Track const&, PipelineStatus);
void update_buffering_state();
void check_for_duration_change(AK::Duration);
void dispatch_error(DecoderError&&);
@ -180,7 +182,9 @@ private:
AK::Duration m_duration;
Optional<AK::UnixDateTime> m_start_time_realtime;
HashTable<Track> m_tracks_still_buffering;
bool m_audio_buffering { false };
HashTable<Track> m_video_tracks_buffering;
bool m_was_buffering { false };
bool m_is_in_error_state { false };
};

3
Libraries/LibMedia/PlaybackStates/SeekingStateHandler.h
View file

@ -179,6 +179,9 @@ private:
m_video_seeks_pending.clear();
m_audio_seeks_pending.clear();
if (manager().m_audio_sink)
manager().m_audio_sink->pause_audio_processor();
for (auto const& track : manager().video_tracks())
start_video_seek(track);

2
Libraries/LibMedia/PlaybackStates/StartingStateHandler.cpp
View file

@ -15,7 +15,7 @@ void StartingStateHandler::start()
{
m_started = true;
if (manager().m_tracks_still_buffering.is_empty())
if (!manager().m_audio_buffering && manager().m_video_tracks_buffering.is_empty())
resume();
}

217
Libraries/LibMedia/Processors/AudioMixer.cpp
View file

@ -16,10 +16,7 @@ ErrorOr<NonnullRefPtr<AudioMixer>> AudioMixer::try_create()
return adopt_nonnull_ref_or_enomem(new (nothrow) AudioMixer);
}
AudioMixer::AudioMixer()
: m_main_thread_event_loop(Core::EventLoop::current())
{
}
AudioMixer::AudioMixer() = default;
void AudioMixer::set_producer(Track const& track, RefPtr<DecodedAudioProducer> const& producer)
{
@ -65,11 +62,13 @@ Audio::SampleSpecification AudioMixer::sample_specification() const
void AudioMixer::reset_to_sample_position(i64 sample_position)
{
m_next_sample_to_write = sample_position;
for (auto& [track, track_data] : m_track_mixing_datas)
for (auto& [track, track_data] : m_track_mixing_datas) {
track_data.current_block.clear();
track_data.last_status = PipelineStatus::Pending;
}
}
bool AudioMixer::mix_one_block_into(AudioBlock& out_block)
PipelineStatus AudioMixer::pull(AudioBlock& into)
{
VERIFY(m_sample_specification.is_valid());
@ -78,146 +77,110 @@ bool AudioMixer::mix_one_block_into(AudioBlock& out_block)
Sync::MutexLocker locker { m_mutex };
auto buffer_start = m_next_sample_to_write;
auto initial_samples_end = buffer_start + static_cast<i64>(max_sample_count);
auto samples_end = initial_samples_end;
auto samples_end_cap = buffer_start + static_cast<i64>(max_sample_count);
auto write_size = max_sample_count * channel_count;
auto buffering = false;
auto any_track_has_fresh_data = false;
for (auto& [track, track_data] : m_track_mixing_datas) {
auto available_end = track_data.producer->queue_end_sample();
// A newly-enabled track has no data at the current mix position yet; skip it for clamping so
// the mixer doesn't stall waiting for it to catch up.
if (available_end <= buffer_start) {
track_data.current_block.clear();
while (true) {
auto block = track_data.producer->retrieve_block();
if (block.is_empty())
break;
if (block.end_timestamp_in_samples() >= buffer_start) {
available_end = block.end_timestamp_in_samples();
track_data.current_block = move(block);
break;
}
}
if (track_data.current_block.is_empty())
continue;
}
any_track_has_fresh_data = true;
if (available_end < samples_end) {
samples_end = available_end;
if (track_data.producer->is_blocked())
buffering = true;
}
}
auto combined_status_after_mix = PipelineStatus::EndOfStream;
i64 latest_mixed_sample = samples_end_cap;
if (!m_track_mixing_datas.is_empty() && !any_track_has_fresh_data)
return false;
for (auto& [track, track_data] : m_track_mixing_datas)
track_data.next_sample = buffer_start;
for (auto& [track, track_data] : m_track_mixing_datas) {
if (!buffering) {
track_data.buffering = false;
} else {
if (!track_data.producer->is_blocked())
continue;
if (track_data.buffering)
continue;
track_data.buffering = true;
m_main_thread_event_loop.deferred_invoke([self = NonnullRefPtr(*this), track] {
if (self->on_start_buffering)
self->on_start_buffering(track);
});
}
}
auto sample_count = static_cast<size_t>(max(samples_end - buffer_start, 0));
auto write_size = sample_count * channel_count;
if (sample_count == 0)
return false;
out_block.emplace(m_sample_specification, buffer_start, [&](AudioBlock::Data& data) {
into.emplace(m_sample_specification, buffer_start, [&](AudioBlock::Data& data) {
data.resize_and_keep_capacity(write_size);
for (size_t i = 0; i < write_size; i++)
data[i] = 0.0f;
for (auto& [track, track_data] : m_track_mixing_datas) {
auto next_sample = buffer_start;
while (true) {
TrackMixingData* next_mix_target = nullptr;
for (auto& [track, track_data] : m_track_mixing_datas) {
if (track_data.next_sample >= samples_end_cap)
continue;
if (next_mix_target == nullptr || track_data.next_sample < next_mix_target->next_sample)
next_mix_target = &track_data;
}
if (next_mix_target == nullptr)
break;
auto go_to_next_block = [&] {
auto new_block = track_data.producer->retrieve_block();
if (new_block.is_empty())
auto& current_block = next_mix_target->current_block;
auto current_block_is_usable = [&] {
if (current_block.is_empty())
return false;
if (current_block.sample_specification() != m_sample_specification)
return false;
if (current_block.end_timestamp_in_samples() <= next_mix_target->next_sample)
return false;
track_data.current_block = move(new_block);
return true;
};
}();
if (track_data.current_block.is_empty()) {
if (!go_to_next_block())
if (!current_block_is_usable) {
current_block.clear();
AudioBlock new_block;
next_mix_target->last_status = next_mix_target->producer->pull(new_block);
if (next_mix_target->last_status == PipelineStatus::EndOfStream) {
next_mix_target->next_sample = samples_end_cap;
continue;
}
if (next_mix_target->last_status != PipelineStatus::HaveData)
break;
VERIFY(!new_block.is_empty());
current_block = move(new_block);
continue;
}
while (!track_data.current_block.is_empty()) {
auto& current_block = track_data.current_block;
auto current_block_sample_count = static_cast<i64>(current_block.sample_count());
if (current_block.sample_specification() != m_sample_specification) {
if (!go_to_next_block())
break;
current_block.clear();
continue;
}
auto first_sample_offset = current_block.timestamp_in_samples();
if (first_sample_offset >= samples_end)
break;
auto block_end = first_sample_offset + current_block_sample_count;
if (block_end <= next_sample) {
if (!go_to_next_block())
break;
continue;
}
next_sample = max(next_sample, first_sample_offset);
VERIFY(next_sample >= first_sample_offset);
auto index_in_block = static_cast<size_t>((next_sample - first_sample_offset) * channel_count);
VERIFY(index_in_block < current_block.data_count());
VERIFY(next_sample >= buffer_start);
auto index_in_buffer = static_cast<size_t>((next_sample - buffer_start) * channel_count);
VERIFY(index_in_buffer < write_size);
VERIFY(current_block.data_count() >= index_in_block);
auto write_count = current_block.data_count() - index_in_block;
write_count = min(write_count, write_size - index_in_buffer);
VERIFY(write_count > 0);
VERIFY(index_in_buffer + write_count <= write_size);
VERIFY(write_count % channel_count == 0);
for (size_t i = 0; i < write_count; i++)
data[index_in_buffer + i] += current_block.data()[index_in_block + i];
auto write_end = index_in_block + write_count;
if (write_end == current_block.data_count()) {
if (!go_to_next_block())
break;
continue;
}
VERIFY(write_end < current_block.data_count());
next_sample += static_cast<i64>(write_count / channel_count);
if (next_sample == samples_end)
break;
VERIFY(next_sample < samples_end);
auto first_sample_offset = current_block.timestamp_in_samples();
if (first_sample_offset >= samples_end_cap) {
next_mix_target->next_sample = samples_end_cap;
continue;
}
auto next_sample = max(next_mix_target->next_sample, first_sample_offset);
VERIFY(next_sample >= first_sample_offset);
auto index_in_block = static_cast<size_t>((next_sample - first_sample_offset) * channel_count);
VERIFY(index_in_block < current_block.data_count());
VERIFY(next_sample >= buffer_start);
auto index_in_buffer = static_cast<size_t>((next_sample - buffer_start) * channel_count);
VERIFY(index_in_buffer < write_size);
VERIFY(current_block.data_count() >= index_in_block);
auto write_count = current_block.data_count() - index_in_block;
write_count = min(write_count, write_size - index_in_buffer);
VERIFY(write_count > 0);
VERIFY(index_in_buffer + write_count <= write_size);
VERIFY(write_count % channel_count == 0);
for (size_t i = 0; i < write_count; i++)
data[index_in_buffer + i] += current_block.data()[index_in_block + i];
next_mix_target->next_sample = next_sample + static_cast<i64>(write_count / channel_count);
}
for (auto& [track, track_data] : m_track_mixing_datas) {
latest_mixed_sample = min(latest_mixed_sample, track_data.next_sample);
combined_status_after_mix = select_combined_pipeline_status(combined_status_after_mix, track_data.last_status);
}
});
VERIFY(latest_mixed_sample >= buffer_start);
auto sample_count = static_cast<size_t>(latest_mixed_sample - buffer_start);
if (combined_status_after_mix == PipelineStatus::EndOfStream) {
m_next_sample_to_write = samples_end_cap;
return PipelineStatus::EndOfStream;
}
if (sample_count == 0) {
into.clear();
if (combined_status_after_mix == PipelineStatus::HaveData)
return PipelineStatus::Pending;
return combined_status_after_mix;
}
into.trim(sample_count);
m_next_sample_to_write += static_cast<i64>(sample_count);
return true;
return PipelineStatus::HaveData;
}
}

12
Libraries/LibMedia/Processors/AudioMixer.h
View file

@ -7,16 +7,15 @@
#pragma once
#include <AK/Atomic.h>
#include <AK/Function.h>
#include <AK/HashMap.h>
#include <AK/NonnullRefPtr.h>
#include <AK/RefPtr.h>
#include <LibCore/EventLoop.h>
#include <LibMedia/Audio/Forward.h>
#include <LibMedia/Audio/SampleSpecification.h>
#include <LibMedia/AudioBlock.h>
#include <LibMedia/Export.h>
#include <LibMedia/Forward.h>
#include <LibMedia/PipelineStatus.h>
#include <LibMedia/Producers/DecodedAudioProducer.h>
#include <LibMedia/Sinks/AudioSink.h>
#include <LibMedia/Track.h>
@ -36,12 +35,10 @@ public:
void set_sample_specification(Audio::SampleSpecification);
Audio::SampleSpecification sample_specification() const;
bool mix_one_block_into(AudioBlock& out_block);
PipelineStatus pull(AudioBlock& into);
void reset_to_sample_position(i64 sample_position);
Function<void(Track const&)> on_start_buffering;
private:
struct TrackMixingData {
TrackMixingData(NonnullRefPtr<DecodedAudioProducer> const& producer)
@ -51,11 +48,10 @@ private:
NonnullRefPtr<DecodedAudioProducer> producer;
AudioBlock current_block;
bool buffering { false };
i64 next_sample { 0 };
PipelineStatus last_status { PipelineStatus::Pending };
};
Core::EventLoop& m_main_thread_event_loop;
mutable Sync::Mutex m_mutex;
Audio::SampleSpecification m_sample_specification;
HashMap<Track, TrackMixingData> m_track_mixing_datas;

4
Libraries/LibMedia/Producers/AudioProducer.h
View file

@ -6,13 +6,17 @@
#pragma once
#include <LibMedia/AudioBlock.h>
#include <LibMedia/MediaPipelineNode.h>
#include <LibMedia/PipelineStatus.h>
namespace Media {
class AudioProducer : public virtual MediaPipelineNode {
public:
virtual ~AudioProducer() = default;
virtual PipelineStatus pull(AudioBlock& into) = 0;
};
}

128
Libraries/LibMedia/Producers/DecodedAudioProducer.cpp
View file

@ -64,11 +64,6 @@ void DecodedAudioProducer::set_duration_change_handler(BlockEndTimeHandler&& han
m_thread_data->set_duration_change_handler(move(handler));
}
void DecodedAudioProducer::set_queue_is_full_handler(QueueIsFullHandler&& handler)
{
m_thread_data->set_queue_is_full_handler(move(handler));
}
void DecodedAudioProducer::set_output_sample_specification(Audio::SampleSpecification sample_specification)
{
m_thread_data->set_output_sample_specification(sample_specification);
@ -94,32 +89,11 @@ void DecodedAudioProducer::seek(AK::Duration timestamp, SeekCompletionHandler&&
m_thread_data->seek(timestamp, move(completion_handler));
}
bool DecodedAudioProducer::is_blocked() const
{
return m_thread_data->is_blocked();
}
i64 DecodedAudioProducer::queue_end_sample() const
{
return m_thread_data->queue_end_sample();
}
TimeRanges DecodedAudioProducer::buffered_time_ranges() const
{
return m_thread_data->buffered_time_ranges();
}
bool DecodedAudioProducer::ThreadData::is_blocked() const
{
return m_demuxer->is_read_blocked_for_track(m_track);
}
i64 DecodedAudioProducer::ThreadData::queue_end_sample() const
{
auto locker = take_lock();
return m_queue_end_sample;
}
DecodedAudioProducer::ThreadData::ThreadData(NonnullRefPtr<Core::WeakEventLoopReference> const& main_thread_event_loop, NonnullRefPtr<Demuxer> const& demuxer, Track const& track, AK::Duration duration, NonnullOwnPtr<Audio::AudioConverter>&& converter)
: m_main_thread_event_loop(main_thread_event_loop)
, m_demuxer(demuxer)
@ -141,11 +115,6 @@ void DecodedAudioProducer::ThreadData::set_duration_change_handler(BlockEndTimeH
m_duration_change_handler = move(handler);
}
void DecodedAudioProducer::ThreadData::set_queue_is_full_handler(QueueIsFullHandler&& handler)
{
m_queue_is_full_handler = move(handler);
}
void DecodedAudioProducer::ThreadData::set_output_sample_specification(Audio::SampleSpecification sample_specification)
{
m_converter->set_output_sample_specification(sample_specification).release_value_but_fixme_should_propagate_errors();
@ -192,14 +161,38 @@ void DecodedAudioProducer::ThreadData::exit()
wake();
}
AudioBlock DecodedAudioProducer::retrieve_block()
PipelineStatus DecodedAudioProducer::pull(AudioBlock& into)
{
auto locker = m_thread_data->take_lock();
if (m_thread_data->queue().is_empty())
return AudioBlock();
auto result = m_thread_data->queue().dequeue();
m_thread_data->wake();
return result;
return m_thread_data->pull(into);
}
PipelineStatus DecodedAudioProducer::ThreadData::pull(AudioBlock& into)
{
auto locker = take_lock();
if (!m_queue.is_empty()) {
into = m_queue.dequeue();
wake();
return PipelineStatus::HaveData;
}
if (m_pending_halting_status != PipelineStatus::Pending)
return m_pending_halting_status;
if (m_demuxer->is_read_blocked_for_track(m_track))
return PipelineStatus::Blocked;
return PipelineStatus::Pending;
}
void DecodedAudioProducer::ThreadData::enter_halting_state(PipelineStatus status, Optional<DecoderError> error)
{
if (error.has_value() && error->category() == DecoderErrorCategory::Aborted)
return;
VERIFY(status == PipelineStatus::EndOfStream || status == PipelineStatus::Error);
m_pending_halting_status = status;
if (error.has_value()) {
invoke_on_main_thread_while_locked([error = error.release_value()](auto const& self) mutable {
self->dispatch_error(move(error));
});
}
}
void DecodedAudioProducer::ThreadData::seek(AK::Duration timestamp, SeekCompletionHandler&& completion_handler)
@ -248,12 +241,8 @@ bool DecodedAudioProducer::ThreadData::handle_suspension()
return true;
auto result = create_decoder();
if (result.is_error()) {
m_is_in_error_state = true;
invoke_on_main_thread_while_locked([error = result.release_error()](auto const& self) mutable {
self->dispatch_error(move(error));
});
}
if (result.is_error())
enter_halting_state(PipelineStatus::Error, result.release_error());
}
// Suspension must be woken with a seek, or we will throw decoding errors.
@ -302,7 +291,6 @@ void DecodedAudioProducer::ThreadData::dispatch_block_end_time(AudioBlock const&
void DecodedAudioProducer::ThreadData::clear_queue()
{
m_queue.clear();
m_queue_end_sample = 0;
}
void DecodedAudioProducer::ThreadData::queue_block(AudioBlock&& block)
@ -311,7 +299,6 @@ void DecodedAudioProducer::ThreadData::queue_block(AudioBlock&& block)
VERIFY(!block.is_empty());
dispatch_block_end_time(block);
m_queue_end_sample = block.end_timestamp_in_samples();
m_queue.enqueue(move(block));
VERIFY(!m_queue.tail().is_empty());
}
@ -357,7 +344,7 @@ void DecodedAudioProducer::ThreadData::process_seek_on_main_thread(u32 seek_id,
void DecodedAudioProducer::ThreadData::resolve_seek(u32 seek_id)
{
m_is_in_error_state = false;
m_pending_halting_status = PipelineStatus::Pending;
process_seek_on_main_thread(seek_id, [](auto& self) {
auto handler = move(self->m_seek_completion_handler);
if (handler)
@ -374,16 +361,12 @@ bool DecodedAudioProducer::ThreadData::handle_seek()
return false;
auto handle_error = [&](DecoderError&& error) {
m_is_in_error_state = true;
{
auto locker = take_lock();
clear_queue();
process_seek_on_main_thread(seek_id,
[error = move(error)](auto& self) mutable {
self->dispatch_error(move(error));
self->m_seek_completion_handler = nullptr;
});
}
auto locker = take_lock();
clear_queue();
enter_halting_state(PipelineStatus::Error, move(error));
process_seek_on_main_thread(seek_id, [](auto& self) {
self->m_seek_completion_handler = nullptr;
});
};
AK::Duration timestamp;
@ -474,17 +457,19 @@ void DecodedAudioProducer::ThreadData::push_data_and_decode_a_block()
{
VERIFY(m_decoder);
auto set_error_and_wait_for_seek = [this](DecoderError&& error) {
auto set_halting_status_and_wait_for_seek = [this](PipelineStatus status, Optional<DecoderError> error) {
{
auto locker = take_lock();
m_is_in_error_state = true;
invoke_on_main_thread_while_locked([error = move(error)](auto const& self) mutable {
self->dispatch_error(move(error));
});
enter_halting_state(status, move(error));
}
dbgln_if(PLAYBACK_MANAGER_DEBUG, "Decoded Audio Producer: Encountered an error, waiting for a seek to start decoding again...");
while (m_is_in_error_state) {
dbgln_if(PLAYBACK_MANAGER_DEBUG, "Decoded Audio Producer: Reached a halting pull status, waiting for a seek to start decoding again...");
while (true) {
{
auto locker = take_lock();
if (m_pending_halting_status == PipelineStatus::Pending)
return;
}
if (handle_seek())
break;
{
@ -501,14 +486,14 @@ void DecodedAudioProducer::ThreadData::push_data_and_decode_a_block()
if (sample_result.error().category() == DecoderErrorCategory::EndOfStream) {
m_decoder->signal_end_of_stream();
} else {
set_error_and_wait_for_seek(sample_result.release_error());
set_halting_status_and_wait_for_seek(PipelineStatus::Error, sample_result.release_error());
return;
}
} else {
auto sample = sample_result.release_value();
auto decode_result = m_decoder->receive_coded_data(sample.timestamp(), sample.data());
if (decode_result.is_error()) {
set_error_and_wait_for_seek(decode_result.release_error());
set_halting_status_and_wait_for_seek(PipelineStatus::Error, decode_result.release_error());
return;
}
}
@ -520,12 +505,6 @@ void DecodedAudioProducer::ThreadData::push_data_and_decode_a_block()
}();
while (queue_size >= m_queue_max_size) {
if (m_queue_is_full_handler) {
invoke_on_main_thread([](auto const& self) {
self->m_queue_is_full_handler();
});
}
if (handle_seek())
return;
@ -549,7 +528,10 @@ void DecodedAudioProducer::ThreadData::push_data_and_decode_a_block()
if (block_result.is_error()) {
if (block_result.error().category() == DecoderErrorCategory::NeedsMoreInput)
break;
set_error_and_wait_for_seek(block_result.release_error());
if (block_result.error().category() == DecoderErrorCategory::EndOfStream)
set_halting_status_and_wait_for_seek(PipelineStatus::EndOfStream, {});
else
set_halting_status_and_wait_for_seek(PipelineStatus::Error, block_result.release_error());
break;
}

19
Libraries/LibMedia/Producers/DecodedAudioProducer.h
View file

@ -8,6 +8,7 @@
#include <AK/NonnullOwnPtr.h>
#include <AK/NonnullRefPtr.h>
#include <AK/Optional.h>
#include <AK/Queue.h>
#include <AK/Time.h>
#include <LibCore/Forward.h>
@ -37,7 +38,6 @@ public:
using ErrorHandler = Function<void(DecoderError&&)>;
using BlockEndTimeHandler = Function<void(AK::Duration)>;
using SeekCompletionHandler = Function<void()>;
using QueueIsFullHandler = Function<void()>;
static DecoderErrorOr<NonnullRefPtr<DecodedAudioProducer>> try_create(NonnullRefPtr<Core::WeakEventLoopReference> const& main_thread_event_loop, NonnullRefPtr<Demuxer> const& demuxer, Track const& track);
DecodedAudioProducer(NonnullRefPtr<ThreadData> const&);
@ -45,20 +45,16 @@ public:
void set_error_handler(ErrorHandler&&);
void set_duration_change_handler(BlockEndTimeHandler&&);
void set_queue_is_full_handler(QueueIsFullHandler&&);
void set_output_sample_specification(Audio::SampleSpecification);
void start();
void suspend();
void resume();
AudioBlock retrieve_block();
virtual PipelineStatus pull(AudioBlock& into) override;
void seek(AK::Duration timestamp, SeekCompletionHandler&& = nullptr);
bool is_blocked() const;
i64 queue_end_sample() const;
TimeRanges buffered_time_ranges() const;
private:
@ -69,7 +65,6 @@ private:
void set_error_handler(ErrorHandler&&);
void set_duration_change_handler(BlockEndTimeHandler&&);
void set_queue_is_full_handler(QueueIsFullHandler&&);
void set_output_sample_specification(Audio::SampleSpecification);
void start();
@ -96,8 +91,8 @@ private:
void process_seek_on_main_thread(u32 seek_id, Callback);
void resolve_seek(u32 seek_id);
void push_data_and_decode_a_block();
bool is_blocked() const;
i64 queue_end_sample() const;
PipelineStatus pull(AudioBlock& into);
TimeRanges buffered_time_ranges() const;
@ -110,6 +105,8 @@ private:
AudioQueue& queue() { return m_queue; }
void clear_queue();
void enter_halting_state(PipelineStatus, Optional<DecoderError>);
private:
enum class RequestedState : u8 {
None,
@ -136,9 +133,7 @@ private:
AudioQueue m_queue;
BlockEndTimeHandler m_duration_change_handler;
ErrorHandler m_error_handler;
bool m_is_in_error_state { false };
QueueIsFullHandler m_queue_is_full_handler;
i64 m_queue_end_sample { 0 };
PipelineStatus m_pending_halting_status { PipelineStatus::Pending };
u32 m_last_processed_seek_id { 0 };
Atomic<u32> m_seek_id { 0 };

114
Libraries/LibMedia/Producers/DecodedVideoProducer.cpp
View file

@ -76,19 +76,38 @@ void DecodedVideoProducer::resume()
m_thread_data->resume();
}
void DecodedVideoProducer::set_frames_queue_is_full_handler(FramesQueueIsFullHandler&& handler)
PipelineStatus DecodedVideoProducer::pull(RefPtr<VideoFrame>& into)
{
m_thread_data->set_frames_queue_is_full_handler(move(handler));
return m_thread_data->pull(into);
}
RefPtr<VideoFrame> DecodedVideoProducer::retrieve_frame()
PipelineStatus DecodedVideoProducer::ThreadData::pull(RefPtr<VideoFrame>& into)
{
auto locker = m_thread_data->take_lock();
if (m_thread_data->queue().is_empty())
return nullptr;
auto result = m_thread_data->take_frame();
m_thread_data->wake();
return result;
auto locker = take_lock();
if (!m_queue.is_empty()) {
into = m_queue.dequeue();
wake();
return PipelineStatus::HaveData;
}
if (m_pending_halting_status != PipelineStatus::Pending)
return m_pending_halting_status;
if (m_demuxer->is_read_blocked_for_track(m_track))
return PipelineStatus::Blocked;
return PipelineStatus::Pending;
}
void DecodedVideoProducer::ThreadData::enter_halting_state(PipelineStatus status, Optional<DecoderError> error)
{
if (error.has_value() && error->category() == DecoderErrorCategory::Aborted)
return;
VERIFY(status == PipelineStatus::EndOfStream || status == PipelineStatus::Error);
m_pending_halting_status = status;
if (error.has_value()) {
invoke_on_main_thread_while_locked([error = error.release_value()](auto const& self) mutable {
self->dispatch_error(move(error));
});
}
}
void DecodedVideoProducer::seek(AK::Duration timestamp, SeekMode seek_mode, SeekCompletionHandler&& completion_handler)
@ -113,11 +132,6 @@ DecoderErrorOr<void> DecodedVideoProducer::ThreadData::create_decoder()
return {};
}
bool DecodedVideoProducer::is_blocked() const
{
return m_thread_data->is_blocked();
}
TimeRanges DecodedVideoProducer::buffered_time_ranges() const
{
return m_thread_data->buffered_time_ranges();
@ -144,11 +158,6 @@ void DecodedVideoProducer::ThreadData::set_duration_change_handler(FrameEndTimeH
m_duration_change_handler = move(handler);
}
void DecodedVideoProducer::ThreadData::set_frames_queue_is_full_handler(FramesQueueIsFullHandler&& handler)
{
m_frames_queue_is_full_handler = move(handler);
}
void DecodedVideoProducer::ThreadData::suspend()
{
auto locker = take_lock();
@ -177,11 +186,6 @@ DecodedVideoProducer::FrameQueue& DecodedVideoProducer::ThreadData::queue()
return m_queue;
}
NonnullRefPtr<VideoFrame> DecodedVideoProducer::ThreadData::take_frame()
{
return m_queue.dequeue();
}
void DecodedVideoProducer::ThreadData::seek(AK::Duration timestamp, SeekMode seek_mode, SeekCompletionHandler&& completion_handler)
{
auto locker = take_lock();
@ -242,12 +246,8 @@ bool DecodedVideoProducer::ThreadData::handle_suspension()
return true;
auto result = create_decoder();
if (result.is_error()) {
m_is_in_error_state = true;
invoke_on_main_thread_while_locked([error = result.release_error()](auto const& self) mutable {
self->dispatch_error(move(error));
});
}
if (result.is_error())
enter_halting_state(PipelineStatus::Error, result.release_error());
}
// Suspension must be woken with a seek, or we will throw decoding errors.
@ -306,7 +306,7 @@ void DecodedVideoProducer::ThreadData::process_seek_on_main_thread(u32 seek_id,
void DecodedVideoProducer::ThreadData::resolve_seek(u32 seek_id, AK::Duration const& timestamp)
{
m_is_in_error_state = false;
m_pending_halting_status = PipelineStatus::Pending;
process_seek_on_main_thread(seek_id, [timestamp](auto& self) {
auto handler = move(self->m_seek_completion_handler);
if (handler)
@ -323,16 +323,12 @@ bool DecodedVideoProducer::ThreadData::handle_seek()
return false;
auto handle_error = [&](DecoderError&& error) {
m_is_in_error_state = true;
{
auto locker = take_lock();
m_queue.clear();
process_seek_on_main_thread(seek_id,
[error = move(error)](auto& self) mutable {
self->dispatch_error(move(error));
self->m_seek_completion_handler = nullptr;
});
}
auto locker = take_lock();
m_queue.clear();
enter_halting_state(PipelineStatus::Error, move(error));
process_seek_on_main_thread(seek_id, [](auto& self) {
self->m_seek_completion_handler = nullptr;
});
};
AK::Duration timestamp;
@ -476,17 +472,19 @@ void DecodedVideoProducer::ThreadData::push_data_and_decode_some_frames()
// Demuxers currently can't report the next keyframe in a convenient way, so that will need implementing
// before this functionality can exist.
auto set_error_and_wait_for_seek = [this](DecoderError&& error) {
auto set_halting_status_and_wait_for_seek = [this](PipelineStatus status, Optional<DecoderError> error) {
{
auto locker = take_lock();
m_is_in_error_state = true;
invoke_on_main_thread_while_locked([error = move(error)](auto const& self) mutable {
self->dispatch_error(move(error));
});
enter_halting_state(status, move(error));
}
dbgln_if(PLAYBACK_MANAGER_DEBUG, "Decoded Video Producer: Encountered an error, waiting for a seek to start decoding again...");
while (m_is_in_error_state) {
dbgln_if(PLAYBACK_MANAGER_DEBUG, "Decoded Video Producer: Reached a halting pull status, waiting for a seek to start decoding again...");
while (true) {
{
auto locker = take_lock();
if (m_pending_halting_status == PipelineStatus::Pending)
return;
}
if (handle_seek())
break;
{
@ -503,7 +501,7 @@ void DecodedVideoProducer::ThreadData::push_data_and_decode_some_frames()
if (sample_result.error().category() == DecoderErrorCategory::EndOfStream) {
m_decoder->signal_end_of_stream();
} else {
set_error_and_wait_for_seek(sample_result.release_error());
set_halting_status_and_wait_for_seek(PipelineStatus::Error, sample_result.release_error());
return;
}
} else {
@ -512,7 +510,7 @@ void DecodedVideoProducer::ThreadData::push_data_and_decode_some_frames()
auto decode_result = m_decoder->receive_coded_data(coded_frame.timestamp(), coded_frame.duration(), coded_frame.data());
if (decode_result.is_error()) {
set_error_and_wait_for_seek(decode_result.release_error());
set_halting_status_and_wait_for_seek(PipelineStatus::Error, decode_result.release_error());
return;
}
}
@ -522,7 +520,10 @@ void DecodedVideoProducer::ThreadData::push_data_and_decode_some_frames()
if (frame_result.is_error()) {
if (frame_result.error().category() == DecoderErrorCategory::NeedsMoreInput)
break;
set_error_and_wait_for_seek(frame_result.release_error());
if (frame_result.error().category() == DecoderErrorCategory::EndOfStream)
set_halting_status_and_wait_for_seek(PipelineStatus::EndOfStream, {});
else
set_halting_status_and_wait_for_seek(PipelineStatus::Error, frame_result.release_error());
break;
}
@ -535,12 +536,6 @@ void DecodedVideoProducer::ThreadData::push_data_and_decode_some_frames()
}();
while (queue_size >= m_queue_max_size) {
if (m_frames_queue_is_full_handler) {
invoke_on_main_thread([](auto const& self) {
self->m_frames_queue_is_full_handler();
});
}
if (handle_seek())
return;
@ -565,11 +560,6 @@ void DecodedVideoProducer::ThreadData::push_data_and_decode_some_frames()
}
}
bool DecodedVideoProducer::ThreadData::is_blocked() const
{
return m_demuxer->is_read_blocked_for_track(m_track);
}
TimeRanges DecodedVideoProducer::ThreadData::buffered_time_ranges() const
{
return m_demuxer->buffered_time_ranges();

17
Libraries/LibMedia/Producers/DecodedVideoProducer.h
View file

@ -9,6 +9,7 @@
#include <AK/Atomic.h>
#include <AK/Forward.h>
#include <AK/NonnullRefPtr.h>
#include <AK/Optional.h>
#include <AK/Queue.h>
#include <AK/Time.h>
#include <LibCore/Forward.h>
@ -36,7 +37,6 @@ public:
using ErrorHandler = Function<void(DecoderError&&)>;
using FrameEndTimeHandler = Function<void(AK::Duration)>;
using SeekCompletionHandler = Function<void(AK::Duration)>;
using FramesQueueIsFullHandler = Function<void()>;
static DecoderErrorOr<NonnullRefPtr<DecodedVideoProducer>> try_create(NonnullRefPtr<Core::WeakEventLoopReference> const& main_thread_event_loop, NonnullRefPtr<Demuxer> const&, Track const&, RefPtr<MediaTimeProvider> const& = nullptr);
@ -45,18 +45,15 @@ public:
void set_error_handler(ErrorHandler&&);
void set_duration_change_handler(FrameEndTimeHandler&&);
void set_frames_queue_is_full_handler(FramesQueueIsFullHandler&&);
void start();
void suspend();
void resume();
RefPtr<VideoFrame> retrieve_frame();
virtual PipelineStatus pull(RefPtr<VideoFrame>& into) override;
void seek(AK::Duration timestamp, SeekMode, SeekCompletionHandler&& = nullptr);
bool is_blocked() const;
TimeRanges buffered_time_ranges() const;
private:
@ -67,7 +64,6 @@ private:
void set_error_handler(ErrorHandler&&);
void set_duration_change_handler(FrameEndTimeHandler&&);
void set_frames_queue_is_full_handler(FramesQueueIsFullHandler&&);
void start();
DecoderErrorOr<void> create_decoder();
@ -76,7 +72,8 @@ private:
void exit();
FrameQueue& queue();
NonnullRefPtr<VideoFrame> take_frame();
PipelineStatus pull(RefPtr<VideoFrame>& into);
void seek(AK::Duration timestamp, SeekMode, SeekCompletionHandler&&);
@ -96,7 +93,8 @@ private:
void process_seek_on_main_thread(u32 seek_id, Callback);
void resolve_seek(u32 seek_id, AK::Duration const& timestamp);
void push_data_and_decode_some_frames();
bool is_blocked() const;
void enter_halting_state(PipelineStatus, Optional<DecoderError>);
TimeRanges buffered_time_ranges() const;
@ -129,8 +127,7 @@ private:
FrameQueue m_queue;
FrameEndTimeHandler m_duration_change_handler;
ErrorHandler m_error_handler;
bool m_is_in_error_state { false };
FramesQueueIsFullHandler m_frames_queue_is_full_handler;
PipelineStatus m_pending_halting_status { PipelineStatus::Pending };
u32 m_last_processed_seek_id { 0 };
Atomic<u32> m_seek_id { 0 };

4
Libraries/LibMedia/Producers/VideoProducer.h
View file

@ -6,13 +6,17 @@
#pragma once
#include <LibMedia/Forward.h>
#include <LibMedia/MediaPipelineNode.h>
#include <LibMedia/PipelineStatus.h>
namespace Media {
class VideoProducer : public virtual MediaPipelineNode {
public:
virtual ~VideoProducer() = default;
virtual PipelineStatus pull(RefPtr<VideoFrame>& into) = 0;
};
}

88
Libraries/LibMedia/Sinks/AudioPlaybackSink.cpp
View file

@ -8,6 +8,7 @@
#include <AK/Atomic.h>
#include <AK/AtomicRefCounted.h>
#include <AK/Time.h>
#include <LibCore/Forward.h>
#include <LibMedia/Audio/PlaybackStream.h>
#include <LibMedia/AudioBlock.h>
#include <LibMedia/Processors/AudioMixer.h>
@ -23,15 +24,19 @@ static constexpr size_t OUTPUT_BLOCK_QUEUE_CAPACITY = 4;
class AudioPlaybackSink::OutputThreadData : public AtomicRefCounted<OutputThreadData> {
public:
OutputThreadData(NonnullRefPtr<AudioMixer>&& mixer)
OutputThreadData(NonnullRefPtr<AudioMixer>&& mixer, PipelineStateChangeHandler on_state_changed)
: m_mixer(move(mixer))
, m_main_thread_event_loop(Core::EventLoop::current_weak())
, m_on_state_changed(move(on_state_changed))
{
}
ReadonlySpan<float> move_output_to_playback_stream_buffer(Span<float>);
void dispatch_state_if_changed(PipelineStatus);
RefPtr<Audio::PlaybackStream> m_playback_stream;
NonnullRefPtr<AudioMixer> m_mixer;
NonnullRefPtr<Core::WeakEventLoopReference> m_main_thread_event_loop;
mutable Sync::Mutex m_output_mutex;
mutable Sync::ConditionVariable m_output_condition { m_output_mutex };
@ -42,14 +47,19 @@ public:
size_t m_block_count { 0 };
i64 m_next_sample_to_play { 0 };
PipelineStateChangeHandler m_on_state_changed;
PipelineStatus m_last_pull_status { PipelineStatus::Pending };
PipelineStatus m_last_dispatched_status { PipelineStatus::Pending };
i64 m_last_real_data_end_in_samples { 0 };
Atomic<bool> m_pause_writing_audio_data { true };
bool m_filler_is_waiting_in_output_loop { false };
bool m_filler_should_exit { false };
bool m_audio_processor_is_waiting_in_output_loop { false };
bool m_audio_processor_should_exit { false };
};
ErrorOr<NonnullRefPtr<AudioPlaybackSink>> AudioPlaybackSink::try_create(NonnullRefPtr<AudioMixer> mixer)
ErrorOr<NonnullRefPtr<AudioPlaybackSink>> AudioPlaybackSink::try_create(NonnullRefPtr<AudioMixer> mixer, PipelineStateChangeHandler on_state_changed)
{
auto output_thread_data = TRY(adopt_nonnull_ref_or_enomem(new (nothrow) OutputThreadData(move(mixer))));
auto output_thread_data = TRY(adopt_nonnull_ref_or_enomem(new (nothrow) OutputThreadData(move(mixer), move(on_state_changed))));
auto sink = TRY(try_make_ref_counted<AudioPlaybackSink>(output_thread_data));
auto thread = TRY(Threading::Thread::try_create("Audio Processor"sv,
@ -58,10 +68,10 @@ ErrorOr<NonnullRefPtr<AudioPlaybackSink>> AudioPlaybackSink::try_create(NonnullR
size_t tail_index;
{
Sync::MutexLocker locker { output_thread_data->m_output_mutex };
output_thread_data->m_filler_is_waiting_in_output_loop = true;
output_thread_data->m_audio_processor_is_waiting_in_output_loop = true;
output_thread_data->m_output_condition.broadcast();
while (true) {
if (output_thread_data->m_filler_should_exit)
if (output_thread_data->m_audio_processor_should_exit)
break;
if (output_thread_data->m_pause_writing_audio_data) {
output_thread_data->m_output_condition.wait();
@ -73,26 +83,35 @@ ErrorOr<NonnullRefPtr<AudioPlaybackSink>> AudioPlaybackSink::try_create(NonnullR
}
break;
}
if (output_thread_data->m_filler_should_exit)
if (output_thread_data->m_audio_processor_should_exit)
return 0;
if (output_thread_data->m_pause_writing_audio_data)
continue;
output_thread_data->m_filler_is_waiting_in_output_loop = false;
output_thread_data->m_audio_processor_is_waiting_in_output_loop = false;
tail_index = output_thread_data->m_block_tail;
}
if (!output_thread_data->m_mixer->mix_one_block_into(output_thread_data->m_blocks[tail_index]))
continue;
auto& output_block = output_thread_data->m_blocks[tail_index];
output_block.clear();
auto status = output_thread_data->m_mixer->pull(output_block);
{
Sync::MutexLocker locker { output_thread_data->m_output_mutex };
output_thread_data->m_block_tail = (tail_index + 1) % OUTPUT_BLOCK_QUEUE_CAPACITY;
output_thread_data->m_block_count++;
output_thread_data->m_last_pull_status = status;
if (!output_block.is_empty()) {
VERIFY(can_carry_data(status));
output_thread_data->m_block_tail = (tail_index + 1) % OUTPUT_BLOCK_QUEUE_CAPACITY;
output_thread_data->m_block_count++;
if (output_thread_data->m_playback_stream)
output_thread_data->m_playback_stream->notify_data_available();
if (output_thread_data->m_playback_stream)
output_thread_data->m_playback_stream->notify_data_available();
output_thread_data->m_output_condition.broadcast();
if (status == PipelineStatus::HaveData)
output_thread_data->m_last_real_data_end_in_samples = output_block.end_timestamp_in_samples();
if (!can_carry_data(output_thread_data->m_last_dispatched_status))
output_thread_data->dispatch_state_if_changed(status);
}
}
}
}));
@ -114,10 +133,20 @@ AudioPlaybackSink::AudioPlaybackSink(NonnullRefPtr<OutputThreadData> output_thre
AudioPlaybackSink::~AudioPlaybackSink()
{
Sync::MutexLocker locker { m_output_thread_data->m_output_mutex };
m_output_thread_data->m_filler_should_exit = true;
m_output_thread_data->m_on_state_changed = nullptr;
m_output_thread_data->m_audio_processor_should_exit = true;
m_output_thread_data->m_output_condition.broadcast();
}
void AudioPlaybackSink::pause_audio_processor()
{
Sync::MutexLocker locker { m_output_thread_data->m_output_mutex };
m_output_thread_data->m_pause_writing_audio_data.store(true);
m_output_thread_data->m_output_condition.broadcast();
while (!m_output_thread_data->m_audio_processor_is_waiting_in_output_loop)
m_output_thread_data->m_output_condition.wait();
}
void AudioPlaybackSink::create_playback_stream()
{
if (m_started_creating_playback_stream)
@ -203,13 +232,33 @@ ReadonlySpan<float> AudioPlaybackSink::OutputThreadData::move_output_to_playback
}
}
if (samples_written < buffer.size())
if (samples_written < buffer.size()) {
buffer = buffer.trim(samples_written);
if (m_last_pull_status == PipelineStatus::Blocked || m_last_pull_status == PipelineStatus::Error)
dispatch_state_if_changed(m_last_pull_status);
}
if (m_last_pull_status == PipelineStatus::EndOfStream && m_next_sample_to_play >= m_last_real_data_end_in_samples)
dispatch_state_if_changed(PipelineStatus::EndOfStream);
m_output_condition.broadcast();
return buffer;
}
void AudioPlaybackSink::OutputThreadData::dispatch_state_if_changed(PipelineStatus status)
{
if (status == m_last_dispatched_status)
return;
m_last_dispatched_status = status;
if (auto event_loop = m_main_thread_event_loop->take(); event_loop.is_alive()) {
event_loop->deferred_invoke([self = NonnullRefPtr(*this), status] {
Sync::MutexLocker locker { self->m_output_mutex };
if (self->m_on_state_changed)
self->m_on_state_changed(status);
});
}
}
AK::Duration AudioPlaybackSink::current_time() const
{
if (m_temporary_time.has_value())
@ -287,13 +336,14 @@ void AudioPlaybackSink::set_time(AK::Duration time)
{
Sync::MutexLocker output_locker { self->m_output_thread_data->m_output_mutex };
while (!self->m_output_thread_data->m_filler_is_waiting_in_output_loop)
while (!self->m_output_thread_data->m_audio_processor_is_waiting_in_output_loop)
self->m_output_thread_data->m_output_condition.wait();
self->m_output_thread_data->m_block_head = 0;
self->m_output_thread_data->m_block_tail = 0;
self->m_output_thread_data->m_block_count = 0;
self->m_output_thread_data->m_mixer->reset_to_sample_position(seek_target_in_samples);
self->m_output_thread_data->m_last_real_data_end_in_samples = seek_target_in_samples;
self->m_output_thread_data->m_next_sample_to_play = seek_target_in_samples;

7
Libraries/LibMedia/Sinks/AudioPlaybackSink.h
View file

@ -6,6 +6,7 @@
#pragma once
#include <AK/Function.h>
#include <AK/NonnullRefPtr.h>
#include <AK/RefPtr.h>
#include <LibCore/EventLoop.h>
@ -13,6 +14,7 @@
#include <LibMedia/Export.h>
#include <LibMedia/Forward.h>
#include <LibMedia/MediaTimeProvider.h>
#include <LibMedia/PipelineStatus.h>
namespace Media {
@ -21,7 +23,7 @@ private:
class OutputThreadData;
public:
static ErrorOr<NonnullRefPtr<AudioPlaybackSink>> try_create(NonnullRefPtr<AudioMixer>);
static ErrorOr<NonnullRefPtr<AudioPlaybackSink>> try_create(NonnullRefPtr<AudioMixer>, PipelineStateChangeHandler on_state_changed);
AudioPlaybackSink(NonnullRefPtr<OutputThreadData>);
virtual ~AudioPlaybackSink() override;
@ -32,6 +34,9 @@ public:
void set_volume(double);
// FIXME: Temporary stopgap until seeks are passed up the chain to the producers.
void pause_audio_processor();
Function<void(Error&&)> on_audio_output_error;
private:

33
Libraries/LibMedia/Sinks/DisplayingVideoSink.cpp
View file

@ -13,13 +13,14 @@
namespace Media {
ErrorOr<NonnullRefPtr<DisplayingVideoSink>> DisplayingVideoSink::try_create(NonnullRefPtr<MediaTimeProvider> const& time_provider)
ErrorOr<NonnullRefPtr<DisplayingVideoSink>> DisplayingVideoSink::try_create(NonnullRefPtr<MediaTimeProvider> const& time_provider, PipelineStateChangeHandler on_state_changed)
{
return TRY(try_make_ref_counted<DisplayingVideoSink>(time_provider));
return TRY(try_make_ref_counted<DisplayingVideoSink>(time_provider, move(on_state_changed)));
}
DisplayingVideoSink::DisplayingVideoSink(NonnullRefPtr<MediaTimeProvider> const& time_provider)
DisplayingVideoSink::DisplayingVideoSink(NonnullRefPtr<MediaTimeProvider> const& time_provider, PipelineStateChangeHandler on_state_changed)
: m_time_provider(time_provider)
, m_on_state_changed(move(on_state_changed))
{
}
@ -53,6 +54,15 @@ RefPtr<DecodedVideoProducer> DisplayingVideoSink::producer(Track const& track) c
return m_producer;
}
void DisplayingVideoSink::dispatch_state_if_changed(PipelineStatus status)
{
if (status == m_last_dispatched_status)
return;
m_last_dispatched_status = status;
if (m_on_state_changed)
m_on_state_changed(status);
}
DisplayingVideoSinkUpdateResult DisplayingVideoSink::update()
{
if (m_producer == nullptr)
@ -67,20 +77,25 @@ DisplayingVideoSinkUpdateResult DisplayingVideoSink::update()
m_has_new_current_frame = false;
}
auto last_pull_status = PipelineStatus::Pending;
while (true) {
if (!m_next_frame) {
m_next_frame = m_producer->retrieve_frame();
if (!m_next_frame) {
if (m_producer->is_blocked() && m_on_start_buffering)
m_on_start_buffering();
if (m_next_frame == nullptr) {
last_pull_status = m_producer->pull(m_next_frame);
if (last_pull_status != PipelineStatus::HaveData)
break;
}
VERIFY(m_next_frame != nullptr);
}
if (m_next_frame->timestamp() > current_time)
break;
m_current_frame = m_next_frame.release_nonnull();
result = DisplayingVideoSinkUpdateResult::NewFrameAvailable;
}
auto effective_status = last_pull_status;
if (m_next_frame != nullptr)
effective_status = PipelineStatus::HaveData;
dispatch_state_if_changed(effective_status);
return result;
}

12
Libraries/LibMedia/Sinks/DisplayingVideoSink.h
View file

@ -6,10 +6,12 @@
#pragma once
#include <AK/Function.h>
#include <AK/NonnullRefPtr.h>
#include <AK/Time.h>
#include <LibMedia/Export.h>
#include <LibMedia/Forward.h>
#include <LibMedia/PipelineStatus.h>
#include <LibMedia/Sinks/VideoSink.h>
#include <LibMedia/Track.h>
@ -22,9 +24,9 @@ enum class DisplayingVideoSinkUpdateResult : u8 {
class MEDIA_API DisplayingVideoSink final : public VideoSink {
public:
static ErrorOr<NonnullRefPtr<DisplayingVideoSink>> try_create(NonnullRefPtr<MediaTimeProvider> const&);
static ErrorOr<NonnullRefPtr<DisplayingVideoSink>> try_create(NonnullRefPtr<MediaTimeProvider> const&, PipelineStateChangeHandler on_state_changed);
DisplayingVideoSink(NonnullRefPtr<MediaTimeProvider> const&);
DisplayingVideoSink(NonnullRefPtr<MediaTimeProvider> const&, PipelineStateChangeHandler);
virtual ~DisplayingVideoSink() override;
void set_time_provider(NonnullRefPtr<MediaTimeProvider> const&);
@ -39,10 +41,9 @@ public:
void pause_updates();
void resume_updates();
Function<void()> m_on_start_buffering;
private:
void verify_track(Track const&) const;
void dispatch_state_if_changed(PipelineStatus);
NonnullRefPtr<MediaTimeProvider> m_time_provider;
RefPtr<DecodedVideoProducer> m_producer;
@ -52,6 +53,9 @@ private:
RefPtr<VideoFrame> m_current_frame;
bool m_pause_updates { false };
bool m_has_new_current_frame { false };
PipelineStateChangeHandler m_on_state_changed;
PipelineStatus m_last_dispatched_status { PipelineStatus::Pending };
};
}

7
Tests/LibMedia/TestDataProducers.cpp
View file

@ -12,6 +12,7 @@
#include <LibMedia/FFmpeg/FFmpegDemuxer.h>
#include <LibMedia/IncrementallyPopulatedStream.h>
#include <LibMedia/MediaTimeProvider.h>
#include <LibMedia/PipelineStatus.h>
#include <LibMedia/Producers/DecodedAudioProducer.h>
#include <LibMedia/Producers/DecodedVideoProducer.h>
#include <LibTest/TestCase.h>
@ -125,8 +126,10 @@ TEST_CASE(audio_producer_underspecified_5_1_channel_map)
auto start_time = MonotonicTime::now_coarse();
while (true) {
auto block = producer->retrieve_block();
if (!block.is_empty()) {
Media::AudioBlock block;
auto status = producer->pull(block);
if (status == Media::PipelineStatus::HaveData) {
EXPECT(!block.is_empty());
EXPECT_EQ(block.channel_count(), 6);
EXPECT_EQ(block.sample_specification().channel_map(), Audio::ChannelMap::surround_5_1());
return;

21
Tests/LibMedia/TestFFmpegAudioNormalization.cpp
View file

@ -11,6 +11,7 @@
#include <LibCore/EventLoop.h>
#include <LibMedia/FFmpeg/FFmpegDemuxer.h>
#include <LibMedia/IncrementallyPopulatedStream.h>
#include <LibMedia/PipelineStatus.h>
#include <LibMedia/Producers/DecodedAudioProducer.h>
#include <LibTest/TestCase.h>
@ -80,12 +81,7 @@ static void decode_and_expect()
VERIFY(!tracks.is_empty());
auto producer = TRY_OR_FAIL(Media::DecodedAudioProducer::try_create(Core::EventLoop::current_weak(), demuxer, tracks[0]));
bool reached_end_of_stream = false;
producer->set_error_handler([&](Media::DecoderError&& error) {
if (error.category() == Media::DecoderErrorCategory::EndOfStream) {
reached_end_of_stream = true;
return;
}
producer->set_error_handler([&](Media::DecoderError&&) {
FAIL("An error occurred while decoding generated WAV data.");
});
producer->start();
@ -93,14 +89,14 @@ static void decode_and_expect()
bool saw_negative_full_scale_sample = false;
bool saw_positive_peak_sample = false;
size_t decoded_sample_count = 0;
bool reached_end_of_stream = false;
MonotonicTime deadline = MonotonicTime::now_coarse() + AK::Duration::from_seconds(1);
while (MonotonicTime::now_coarse() < deadline) {
auto block = producer->retrieve_block();
if (block.is_empty()) {
if (reached_end_of_stream)
break;
} else {
Media::AudioBlock block;
auto status = producer->pull(block);
if (status == Media::PipelineStatus::HaveData) {
EXPECT(!block.is_empty());
for (float sample : block.data()) {
EXPECT(sample >= -1.0f);
if constexpr (sizeof(Sample) >= sizeof(i32))
@ -113,6 +109,9 @@ static void decode_and_expect()
saw_positive_peak_sample = true;
}
decoded_sample_count += block.sample_count();
} else if (status == Media::PipelineStatus::EndOfStream) {
reached_end_of_stream = true;
break;
}
loop.pump(Core::EventLoop::WaitMode::PollForEvents);

21
Tests/LibMedia/TestMediaCommon.h
View file

@ -15,6 +15,7 @@
#include <LibMedia/Containers/Matroska/Reader.h>
#include <LibMedia/Demuxer.h>
#include <LibMedia/FFmpeg/FFmpegDemuxer.h>
#include <LibMedia/PipelineStatus.h>
#include <LibMedia/Producers/DecodedAudioProducer.h>
#include <LibMedia/VideoDecoder.h>
#include <LibMedia/VideoFrame.h>
@ -84,12 +85,7 @@ static inline void decode_audio(StringView path, u32 sample_rate, u8 channel_cou
VERIFY(!tracks.is_empty());
auto producer = TRY_OR_FAIL(Media::DecodedAudioProducer::try_create(Core::EventLoop::current_weak(), demuxer, tracks[0]));
auto reached_end = false;
producer->set_error_handler([&](Media::DecoderError&& error) {
if (error.category() == Media::DecoderErrorCategory::EndOfStream) {
reached_end = true;
return;
}
producer->set_error_handler([&](Media::DecoderError&&) {
FAIL("An error occurred while decoding.");
});
producer->start();
@ -99,13 +95,13 @@ static inline void decode_audio(StringView path, u32 sample_rate, u8 channel_cou
i64 last_sample = 0;
size_t sample_count = 0;
auto reached_end = false;
while (true) {
auto block = producer->retrieve_block();
if (block.is_empty()) {
if (reached_end)
break;
} else {
Media::AudioBlock block;
auto status = producer->pull(block);
if (status == Media::PipelineStatus::HaveData) {
EXPECT(!block.is_empty());
EXPECT_EQ(block.sample_rate(), sample_rate);
EXPECT_EQ(block.channel_count(), channel_count);
if (expected_channel_map.has_value())
@ -115,6 +111,9 @@ static inline void decode_audio(StringView path, u32 sample_rate, u8 channel_cou
last_sample = block.timestamp_in_samples() + static_cast<i64>(block.sample_count());
sample_count += block.sample_count();
} else if (status == Media::PipelineStatus::EndOfStream) {
reached_end = true;
break;
}
if (MonotonicTime::now_coarse() - start_time >= time_limit) {