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

LibMedia: Signal downstream nodes whether to clear data after seeks

Browse source 
Seeks don't always move a decoded data producer's head, so we need to
make sure not to remove queued data downstream when that is the case.

To communicate this, the producers can now be queried before pulling
data, allowing them to have an in-band signal to clear the queued data
after a seek has moved the producer and broken monotonicity.

This fixes a flake in HTMLVideoElement-resize-event-during-playback.
This commit is contained in:
Zaggy1024 2026-05-18 22:41:00 -05:00 committed by Gregory Bertilson
parent 5bd13d061e
commit adb0754bc8
16 changed files with 382 additions and 213 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
Libraries/LibMedia/PipelineStatus.h
View file

@ -15,6 +15,7 @@ namespace Media {
enum class PipelineStatus : u8 {
Pending,
HaveData,
MovedPosition,
Blocked,
EndOfStream,
Error,
@ -24,6 +25,8 @@ constexpr bool is_waiting_for_data(PipelineStatus status)
{
if (status == PipelineStatus::Pending)
return true;
if (status == PipelineStatus::MovedPosition)
return true;
if (status == PipelineStatus::Blocked)
return true;
return false;
@ -46,12 +49,15 @@ constexpr PipelineStatus select_combined_pipeline_status(PipelineStatus a, Pipel
return PipelineStatus::Blocked;
if (a == PipelineStatus::HaveData || b == PipelineStatus::HaveData)
return PipelineStatus::HaveData;
if (a == PipelineStatus::MovedPosition || b == PipelineStatus::MovedPosition)
return PipelineStatus::MovedPosition;
if (a == PipelineStatus::Pending || b == PipelineStatus::Pending)
return PipelineStatus::Pending;
return PipelineStatus::EndOfStream;
}
using PipelineStateChangeHandler = Function<void(PipelineStatus)>;
using PipelineWakeHandler = Function<void()>;
constexpr StringView pipeline_status_to_string(PipelineStatus status)
{
@ -60,6 +66,8 @@ constexpr StringView pipeline_status_to_string(PipelineStatus status)
return "Pending"sv;
case PipelineStatus::HaveData:
return "HaveData"sv;
case PipelineStatus::MovedPosition:
return "MovedPosition"sv;
case PipelineStatus::Blocked:
return "Blocked"sv;
case PipelineStatus::EndOfStream:

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

@ -63,7 +63,7 @@ public:
virtual void on_audio_sink_state_changed(PipelineStatus status) override
{
if (status == PipelineStatus::Pending)
if (is_waiting_for_data(status))
return;
m_audio_seek_pending = false;
possibly_complete_seek();
@ -71,7 +71,7 @@ public:
virtual void on_video_sink_state_changed(Track const& track, PipelineStatus status) override
{
if (status == PipelineStatus::Pending)
if (is_waiting_for_data(status))
return;
m_video_seeks_pending.remove(track);
possibly_complete_seek();

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

@ -5,6 +5,7 @@
*/
#include <LibCore/EventLoop.h>
#include <LibMedia/PipelineStatus.h>
#include <LibMedia/Processors/AudioMixer.h>
#include <LibMedia/Producers/DecodedAudioProducer.h>
@ -23,7 +24,7 @@ AudioMixer::~AudioMixer()
{
Sync::MutexLocker locker { m_mutex };
for (auto& [input, input_data] : m_inputs)
input->set_state_changed_handler(nullptr);
input->set_wake_handler(nullptr);
}
ErrorOr<void> AudioMixer::connect_input(NonnullRefPtr<AudioProducer> const& input)
@ -31,15 +32,28 @@ ErrorOr<void> AudioMixer::connect_input(NonnullRefPtr<AudioProducer> const& inpu
Sync::MutexLocker locker { m_mutex };
VERIFY(!m_inputs.contains(input));
m_inputs.set(input, InputMixingData());
input->set_state_changed_handler([this, input](PipelineStatus status) {
Sync::MutexLocker locker { m_mutex };
auto input_data = m_inputs.get(input);
if (!input_data.has_value())
return;
if (input_data->last_status == status)
return;
input_data->last_status = status;
dispatch_state_if_changed(combined_input_status());
input->set_wake_handler([this, input] {
auto input_status = input->status();
bool should_wake_downstream;
{
Sync::MutexLocker locker { m_mutex };
auto& input_data = m_inputs.get(input).value();
if (input_data.last_status == input_status)
return;
input_data.last_status = input_status;
while (input_data.last_status == PipelineStatus::MovedPosition) {
input->pull(input_data.current_block);
VERIFY(input_data.current_block.is_empty());
input_data.last_status = input->status();
}
m_status = combined_input_status();
should_wake_downstream = m_downstream_needs_wake && !is_waiting_for_data(m_status);
if (m_moved_position_pending)
m_status = PipelineStatus::MovedPosition;
}
if (should_wake_downstream)
dispatch_wake();
});
if (m_sample_specification.is_valid()) {
if (auto result = input->set_output_sample_specification(m_sample_specification); result.is_error()) {
@ -62,9 +76,16 @@ void AudioMixer::disconnect_input(NonnullRefPtr<AudioProducer> const& input)
void AudioMixer::disconnect_input_while_locked(NonnullRefPtr<AudioProducer> const& input)
{
input->set_state_changed_handler(nullptr);
input->set_wake_handler(nullptr);
m_inputs.remove(input);
dispatch_state_if_changed(combined_input_status());
m_status = combined_input_status();
if (m_downstream_needs_wake && !is_waiting_for_data(m_status)) {
Core::deferred_invoke([self = NonnullRefPtr(*this)] {
self->dispatch_wake();
});
}
if (m_moved_position_pending)
m_status = PipelineStatus::MovedPosition;
}
ErrorOr<void> AudioMixer::set_output_sample_specification(Audio::SampleSpecification sample_specification)
@ -120,19 +141,26 @@ void AudioMixer::seek(AK::Duration timestamp)
if (!m_sample_specification.is_valid())
return;
m_moved_position_pending = true;
m_next_frame_to_write = timestamp.to_time_units(1, m_sample_specification.sample_rate());
for (auto& [input, input_data] : m_inputs) {
input_data.current_block.clear();
input_data.next_frame = m_next_frame_to_write;
input_data.last_status = PipelineStatus::Pending;
}
m_last_dispatched_status = PipelineStatus::Pending;
m_status = PipelineStatus::Pending;
m_downstream_needs_wake = true;
}
if (m_inputs.is_empty()) {
Core::deferred_invoke([self = NonnullRefPtr(*this)] {
self->dispatch_state_if_changed(PipelineStatus::EndOfStream);
{
Sync::MutexLocker locker { self->m_mutex };
self->m_status = PipelineStatus::MovedPosition;
}
self->dispatch_wake();
});
return;
}
@ -141,18 +169,24 @@ void AudioMixer::seek(AK::Duration timestamp)
input->seek(timestamp);
}
void AudioMixer::set_state_changed_handler(PipelineStateChangeHandler handler)
PipelineStatus AudioMixer::status() const
{
m_state_changed_handler = move(handler);
Sync::MutexLocker locker { m_mutex };
m_downstream_needs_wake = is_waiting_for_data(m_status);
return m_status;
}
void AudioMixer::dispatch_state_if_changed(PipelineStatus status)
void AudioMixer::set_wake_handler(PipelineWakeHandler handler)
{
if (m_last_dispatched_status == status)
return;
m_last_dispatched_status = status;
if (m_state_changed_handler)
m_state_changed_handler(status);
m_wake_handler = move(handler);
}
void AudioMixer::dispatch_wake()
{
if (m_wake_handler)
m_wake_handler();
Sync::MutexLocker locker { m_mutex };
m_downstream_needs_wake = false;
}
PipelineStatus AudioMixer::combined_input_status() const
@ -160,10 +194,11 @@ PipelineStatus AudioMixer::combined_input_status() const
auto status = PipelineStatus::EndOfStream;
for (auto const& [input, input_data] : m_inputs)
status = select_combined_pipeline_status(status, input_data.last_status);
VERIFY(status != PipelineStatus::MovedPosition);
return status;
}
PipelineStatus AudioMixer::pull(AudioBlock& into)
void AudioMixer::pull(AudioBlock& into)
{
VERIFY(m_sample_specification.is_valid());
@ -171,6 +206,13 @@ PipelineStatus AudioMixer::pull(AudioBlock& into)
auto max_frame_count = MAX_SAMPLES_PER_OUTPUT_BLOCK / channel_count;
Sync::MutexLocker locker { m_mutex };
if (m_moved_position_pending) {
m_moved_position_pending = false;
into.clear();
m_status = combined_input_status();
return;
}
auto buffer_start_frame = m_next_frame_to_write;
auto frames_end_cap = buffer_start_frame + static_cast<i64>(max_frame_count);
auto write_size = max_frame_count * channel_count;
@ -206,6 +248,13 @@ PipelineStatus AudioMixer::pull(AudioBlock& into)
auto [input, input_data] = mix_target.release_value();
auto& current_block = input_data.current_block;
input_data.last_status = input.status();
while (input_data.last_status == PipelineStatus::MovedPosition) {
input.pull(current_block);
VERIFY(current_block.is_empty());
input_data.last_status = input.status();
}
auto current_block_is_usable = [&] {
if (current_block.is_empty())
return false;
@ -218,16 +267,14 @@ PipelineStatus AudioMixer::pull(AudioBlock& into)
if (!current_block_is_usable) {
current_block.clear();
AudioBlock new_block;
input_data.last_status = input.pull(new_block);
if (input_data.last_status == PipelineStatus::EndOfStream) {
input_data.next_frame = frames_end_cap;
continue;
}
if (input_data.last_status != PipelineStatus::HaveData)
break;
VERIFY(!new_block.is_empty());
current_block = move(new_block);
input.pull(current_block);
VERIFY(!current_block.is_empty());
continue;
}
@ -261,6 +308,7 @@ PipelineStatus AudioMixer::pull(AudioBlock& into)
}
for (auto& [input, input_data] : m_inputs) {
VERIFY(input_data.last_status != PipelineStatus::MovedPosition);
latest_mixed_frame = min(latest_mixed_frame, input_data.next_frame);
combined_status_after_mix = select_combined_pipeline_status(combined_status_after_mix, input_data.last_status);
}
@ -271,19 +319,22 @@ PipelineStatus AudioMixer::pull(AudioBlock& into)
if (combined_status_after_mix == PipelineStatus::EndOfStream) {
m_next_frame_to_write = frames_end_cap;
return PipelineStatus::EndOfStream;
m_status = PipelineStatus::EndOfStream;
return;
}
if (frame_count == 0) {
into.clear();
if (combined_status_after_mix == PipelineStatus::HaveData)
return PipelineStatus::Pending;
return combined_status_after_mix;
VERIFY(combined_status_after_mix != PipelineStatus::HaveData);
VERIFY(combined_status_after_mix != PipelineStatus::MovedPosition);
VERIFY(combined_status_after_mix != PipelineStatus::EndOfStream);
m_status = combined_status_after_mix;
return;
}
into.trim(frame_count);
m_next_frame_to_write += static_cast<i64>(frame_count);
return PipelineStatus::HaveData;
m_status = PipelineStatus::HaveData;
}
}

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

@ -38,9 +38,10 @@ public:
virtual void start() override;
virtual PipelineStatus pull(AudioBlock& into) override;
virtual PipelineStatus status() const override;
virtual void pull(AudioBlock& into) override;
virtual void set_state_changed_handler(PipelineStateChangeHandler) override;
virtual void set_wake_handler(PipelineWakeHandler) override;
private:
struct InputMixingData {
@ -49,8 +50,8 @@ private:
PipelineStatus last_status { PipelineStatus::Pending };
};
void dispatch_state_if_changed(PipelineStatus);
PipelineStatus combined_input_status() const;
void dispatch_wake();
AK::Duration mix_head_timestamp() const;
void disconnect_input_while_locked(NonnullRefPtr<AudioProducer> const&);
@ -60,9 +61,11 @@ private:
HashMap<NonnullRefPtr<AudioProducer>, InputMixingData> m_inputs;
i64 m_next_frame_to_write { 0 };
bool m_started { false };
bool m_moved_position_pending { false };
mutable bool m_downstream_needs_wake { true };
PipelineStateChangeHandler m_state_changed_handler;
PipelineStatus m_last_dispatched_status { PipelineStatus::Pending };
PipelineWakeHandler m_wake_handler;
PipelineStatus m_status { PipelineStatus::Pending };
};
}

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

@ -21,8 +21,9 @@ public:
virtual void start() = 0;
virtual ErrorOr<void> set_output_sample_specification(Audio::SampleSpecification) = 0;
virtual PipelineStatus pull(AudioBlock& into) = 0;
virtual void set_state_changed_handler(PipelineStateChangeHandler) = 0;
virtual PipelineStatus status() const = 0;
virtual void pull(AudioBlock& into) = 0;
virtual void set_wake_handler(PipelineWakeHandler) = 0;
virtual void seek(AK::Duration timestamp) = 0;
};

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

@ -69,9 +69,14 @@ ErrorOr<void> DecodedAudioProducer::set_output_sample_specification(Audio::Sampl
return m_thread_data->set_output_sample_specification(sample_specification);
}
void DecodedAudioProducer::set_state_changed_handler(PipelineStateChangeHandler handler)
PipelineStatus DecodedAudioProducer::status() const
{
m_thread_data->set_state_changed_handler(move(handler));
return m_thread_data->status();
}
void DecodedAudioProducer::set_wake_handler(PipelineWakeHandler handler)
{
m_thread_data->set_wake_handler(move(handler));
}
void DecodedAudioProducer::start()
@ -126,24 +131,24 @@ ErrorOr<void> DecodedAudioProducer::ThreadData::set_output_sample_specification(
return {};
}
void DecodedAudioProducer::ThreadData::set_state_changed_handler(PipelineStateChangeHandler handler)
void DecodedAudioProducer::ThreadData::set_wake_handler(PipelineWakeHandler handler)
{
auto locker = take_lock();
m_state_changed_handler = move(handler);
m_wake_handler = move(handler);
}
void DecodedAudioProducer::ThreadData::dispatch_state_if_changed_while_locked(PipelineStatus status)
void DecodedAudioProducer::ThreadData::dispatch_wake_if_needed_while_locked()
{
if (status == m_last_dispatched_status)
if (!m_downstream_needs_wake)
return;
m_last_dispatched_status = status;
auto seek_id = m_seek_id.load();
invoke_on_main_thread_while_locked([status, seek_id](auto& self) {
invoke_on_main_thread_while_locked([seek_id](auto& self) {
if (self->m_seek_id != seek_id)
return;
if (self->m_state_changed_handler)
self->m_state_changed_handler(status);
if (self->m_wake_handler)
self->m_wake_handler();
});
m_downstream_needs_wake = false;
}
void DecodedAudioProducer::ThreadData::start()
@ -187,28 +192,48 @@ void DecodedAudioProducer::ThreadData::exit()
wake();
}
PipelineStatus DecodedAudioProducer::pull(AudioBlock& into)
void DecodedAudioProducer::pull(AudioBlock& into)
{
return m_thread_data->pull(into);
m_thread_data->pull(into);
}
PipelineStatus DecodedAudioProducer::ThreadData::pull(AudioBlock& into)
PipelineStatus DecodedAudioProducer::ThreadData::status() const
{
auto locker = take_lock();
auto status = status_while_locked();
m_downstream_needs_wake = is_waiting_for_data(status);
return status;
}
PipelineStatus DecodedAudioProducer::ThreadData::status_while_locked() const
{
if (m_last_processed_seek_id != m_seek_id)
return PipelineStatus::Pending;
if (m_moved_position_pending)
return PipelineStatus::MovedPosition;
if (!m_queue.is_empty())
return PipelineStatus::HaveData;
if (m_current_halting_status != PipelineStatus::Pending)
return m_current_halting_status;
if (m_demuxer->is_read_blocked_for_track(m_track))
return PipelineStatus::Blocked;
return PipelineStatus::Pending;
}
void DecodedAudioProducer::ThreadData::pull(AudioBlock& into)
{
auto locker = take_lock();
if (m_moved_position_pending) {
m_moved_position_pending = false;
into.clear();
return;
}
if (!m_queue.is_empty()) {
into = m_queue.dequeue();
wake();
return PipelineStatus::HaveData;
return;
}
auto status = [&] {
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;
}();
dispatch_state_if_changed_while_locked(status);
return status;
into.clear();
}
void DecodedAudioProducer::ThreadData::enter_halting_state(PipelineStatus status, Optional<DecoderError> error)
@ -217,8 +242,8 @@ void DecodedAudioProducer::ThreadData::enter_halting_state(PipelineStatus status
return;
VERIFY(status == PipelineStatus::EndOfStream || status == PipelineStatus::Error);
m_pending_halting_status = status;
dispatch_state_if_changed_while_locked(status);
m_current_halting_status = status;
dispatch_wake_if_needed_while_locked();
if (error.has_value()) {
invoke_on_main_thread_while_locked([error = error.release_value()](auto const& self) mutable {
self->dispatch_error(move(error));
@ -231,10 +256,9 @@ void DecodedAudioProducer::ThreadData::seek(AK::Duration timestamp)
auto locker = take_lock();
m_seek_id++;
m_seek_timestamp = timestamp;
clear_queue();
m_pending_halting_status = PipelineStatus::Pending;
m_current_halting_status = PipelineStatus::Pending;
m_downstream_needs_wake = true;
m_demuxer->set_blocking_reads_aborted_for_track(m_track);
dispatch_state_if_changed_while_locked(PipelineStatus::Pending);
wake();
}
@ -336,7 +360,7 @@ void DecodedAudioProducer::ThreadData::queue_block(AudioBlock&& block)
dispatch_block_end_time(block);
m_queue.enqueue(move(block));
VERIFY(!m_queue.tail().is_empty());
dispatch_state_if_changed_while_locked(PipelineStatus::HaveData);
dispatch_wake_if_needed_while_locked();
}
void DecodedAudioProducer::ThreadData::dispatch_error(DecoderError&& error)
@ -367,10 +391,12 @@ DecoderErrorOr<void> DecodedAudioProducer::ThreadData::retrieve_next_block(Audio
return {};
}
void DecodedAudioProducer::ThreadData::resolve_seek(u32 seek_id)
void DecodedAudioProducer::ThreadData::resolve_seek(u32 seek_id, bool moved_position)
{
clear_queue();
m_last_processed_seek_id = seek_id;
VERIFY(m_pending_halting_status != PipelineStatus::HaveData);
VERIFY(m_current_halting_status != PipelineStatus::HaveData);
m_moved_position_pending = moved_position;
}
bool DecodedAudioProducer::ThreadData::handle_seek()
@ -389,6 +415,7 @@ bool DecodedAudioProducer::ThreadData::handle_seek()
};
AK::Duration timestamp;
bool moved_position = false;
while (true) {
{
@ -410,8 +437,10 @@ bool DecodedAudioProducer::ThreadData::handle_seek()
}
auto demuxer_seek_result = demuxer_seek_result_or_error.value_or(DemuxerSeekResult::MovedPosition);
if (demuxer_seek_result == DemuxerSeekResult::MovedPosition)
if (demuxer_seek_result == DemuxerSeekResult::MovedPosition) {
flush_decoder();
moved_position = true;
}
auto new_seek_id = seek_id;
AudioBlock last_block;
@ -440,7 +469,7 @@ bool DecodedAudioProducer::ThreadData::handle_seek()
if (block_result.is_error()) {
if (block_result.error().category() == DecoderErrorCategory::EndOfStream) {
auto locker = take_lock();
resolve_seek(seek_id);
resolve_seek(seek_id, moved_position);
return true;
}
@ -453,8 +482,7 @@ bool DecodedAudioProducer::ThreadData::handle_seek()
if (current_block.timestamp() > timestamp) {
auto locker = take_lock();
clear_queue();
resolve_seek(seek_id);
resolve_seek(seek_id, moved_position);
if (!last_block.is_empty())
queue_block(move(last_block));
@ -485,7 +513,7 @@ void DecodedAudioProducer::ThreadData::push_data_and_decode_a_block()
while (true) {
{
auto locker = take_lock();
if (m_pending_halting_status == PipelineStatus::Pending)
if (m_current_halting_status == PipelineStatus::Pending)
return;
}
if (handle_seek())

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

@ -50,8 +50,9 @@ public:
void suspend();
void resume();
virtual PipelineStatus pull(AudioBlock& into) override;
virtual void set_state_changed_handler(PipelineStateChangeHandler) override;
virtual PipelineStatus status() const override;
virtual void pull(AudioBlock& into) override;
virtual void set_wake_handler(PipelineWakeHandler) override;
virtual void seek(AK::Duration timestamp) override;
@ -66,7 +67,7 @@ private:
void set_error_handler(ErrorHandler&&);
void set_duration_change_handler(BlockEndTimeHandler&&);
ErrorOr<void> set_output_sample_specification(Audio::SampleSpecification);
void set_state_changed_handler(PipelineStateChangeHandler);
void set_wake_handler(PipelineWakeHandler);
void start();
DecoderErrorOr<void> create_decoder();
@ -88,10 +89,12 @@ private:
void flush_decoder();
DecoderErrorOr<void> retrieve_next_block(AudioBlock&);
bool handle_seek();
void resolve_seek(u32 seek_id);
void resolve_seek(u32 seek_id, bool moved_position);
void push_data_and_decode_a_block();
PipelineStatus pull(AudioBlock& into);
PipelineStatus status() const;
PipelineStatus status_while_locked() const;
void pull(AudioBlock& into);
TimeRanges buffered_time_ranges() const;
@ -104,7 +107,7 @@ private:
AudioQueue& queue() { return m_queue; }
void clear_queue();
void dispatch_state_if_changed_while_locked(PipelineStatus);
void dispatch_wake_if_needed_while_locked();
void enter_halting_state(PipelineStatus, Optional<DecoderError>);
@ -134,14 +137,15 @@ private:
AudioQueue m_queue;
BlockEndTimeHandler m_duration_change_handler;
ErrorHandler m_error_handler;
PipelineStatus m_pending_halting_status { PipelineStatus::Pending };
PipelineStatus m_current_halting_status { PipelineStatus::Pending };
bool m_moved_position_pending { false };
u32 m_last_processed_seek_id { 0 };
Atomic<u32> m_seek_id { 0 };
AK::Duration m_seek_timestamp;
PipelineStateChangeHandler m_state_changed_handler;
PipelineStatus m_last_dispatched_status { PipelineStatus::Pending };
PipelineWakeHandler m_wake_handler;
mutable bool m_downstream_needs_wake { true };
};
NonnullRefPtr<ThreadData> m_thread_data;

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

@ -76,33 +76,58 @@ void DecodedVideoProducer::resume()
m_thread_data->resume();
}
PipelineStatus DecodedVideoProducer::pull(RefPtr<VideoFrame>& into)
void DecodedVideoProducer::pull(RefPtr<VideoFrame>& into)
{
return m_thread_data->pull(into);
m_thread_data->pull(into);
}
void DecodedVideoProducer::set_state_changed_handler(PipelineStateChangeHandler handler)
PipelineStatus DecodedVideoProducer::status() const
{
m_thread_data->set_state_changed_handler(move(handler));
return m_thread_data->status();
}
PipelineStatus DecodedVideoProducer::ThreadData::pull(RefPtr<VideoFrame>& into)
void DecodedVideoProducer::set_wake_handler(PipelineWakeHandler handler)
{
m_thread_data->set_wake_handler(move(handler));
}
PipelineStatus DecodedVideoProducer::ThreadData::status() const
{
auto locker = take_lock();
auto status = status_while_locked();
m_downstream_needs_wake = is_waiting_for_data(status);
return status;
}
PipelineStatus DecodedVideoProducer::ThreadData::status_while_locked() const
{
if (m_last_processed_seek_id != m_seek_id)
return PipelineStatus::Pending;
if (m_moved_position_pending)
return PipelineStatus::MovedPosition;
if (!m_queue.is_empty())
return PipelineStatus::HaveData;
if (m_current_halting_status != PipelineStatus::Pending)
return m_current_halting_status;
if (m_demuxer->is_read_blocked_for_track(m_track))
return PipelineStatus::Blocked;
return PipelineStatus::Pending;
}
void DecodedVideoProducer::ThreadData::pull(RefPtr<VideoFrame>& into)
{
auto locker = take_lock();
if (m_moved_position_pending) {
m_moved_position_pending = false;
into = nullptr;
return;
}
if (!m_queue.is_empty()) {
into = m_queue.dequeue();
wake();
return PipelineStatus::HaveData;
return;
}
auto status = [&] {
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;
}();
dispatch_state_if_changed_while_locked(status);
return status;
into = nullptr;
}
void DecodedVideoProducer::ThreadData::enter_halting_state(PipelineStatus status, Optional<DecoderError> error)
@ -111,8 +136,8 @@ void DecodedVideoProducer::ThreadData::enter_halting_state(PipelineStatus status
return;
VERIFY(status == PipelineStatus::EndOfStream || status == PipelineStatus::Error);
m_pending_halting_status = status;
dispatch_state_if_changed_while_locked(status);
m_current_halting_status = status;
dispatch_wake_if_needed_while_locked();
if (error.has_value()) {
invoke_on_main_thread_while_locked([error = error.release_value()](auto const& self) mutable {
self->dispatch_error(move(error));
@ -120,24 +145,24 @@ void DecodedVideoProducer::ThreadData::enter_halting_state(PipelineStatus status
}
}
void DecodedVideoProducer::ThreadData::set_state_changed_handler(PipelineStateChangeHandler handler)
void DecodedVideoProducer::ThreadData::set_wake_handler(PipelineWakeHandler handler)
{
auto locker = take_lock();
m_state_changed_handler = move(handler);
m_wake_handler = move(handler);
}
void DecodedVideoProducer::ThreadData::dispatch_state_if_changed_while_locked(PipelineStatus status)
void DecodedVideoProducer::ThreadData::dispatch_wake_if_needed_while_locked()
{
if (status == m_last_dispatched_status)
if (!m_downstream_needs_wake)
return;
m_last_dispatched_status = status;
auto seek_id = m_seek_id.load();
invoke_on_main_thread_while_locked([status, seek_id](auto& self) {
invoke_on_main_thread_while_locked([seek_id](auto& self) {
if (self->m_seek_id != seek_id)
return;
if (self->m_state_changed_handler)
self->m_state_changed_handler(status);
if (self->m_wake_handler)
self->m_wake_handler();
});
m_downstream_needs_wake = false;
}
AK::Duration DecodedVideoProducer::select_fast_seek_target(AK::Duration timestamp, SeekMode mode)
@ -226,10 +251,9 @@ void DecodedVideoProducer::ThreadData::seek(AK::Duration timestamp)
auto locker = take_lock();
m_seek_id++;
m_seek_timestamp = timestamp;
m_queue.clear();
m_pending_halting_status = PipelineStatus::Pending;
m_current_halting_status = PipelineStatus::Pending;
m_downstream_needs_wake = true;
m_demuxer->set_blocking_reads_aborted_for_track(m_track);
dispatch_state_if_changed_while_locked(PipelineStatus::Pending);
wake();
}
@ -326,7 +350,7 @@ void DecodedVideoProducer::ThreadData::queue_frame(NonnullRefPtr<VideoFrame> con
if (m_seek_id.load() != m_last_processed_seek_id)
return;
m_queue.enqueue(frame);
dispatch_state_if_changed_while_locked(PipelineStatus::HaveData);
dispatch_wake_if_needed_while_locked();
}
void DecodedVideoProducer::ThreadData::dispatch_error(DecoderError&& error)
@ -337,10 +361,12 @@ void DecodedVideoProducer::ThreadData::dispatch_error(DecoderError&& error)
m_error_handler(move(error));
}
void DecodedVideoProducer::ThreadData::resolve_seek(u32 seek_id)
void DecodedVideoProducer::ThreadData::resolve_seek(u32 seek_id, bool moved_position)
{
m_queue.clear();
m_last_processed_seek_id = seek_id;
VERIFY(m_pending_halting_status != PipelineStatus::HaveData);
VERIFY(m_current_halting_status != PipelineStatus::HaveData);
m_moved_position_pending = moved_position;
}
bool DecodedVideoProducer::ThreadData::handle_seek()
@ -359,6 +385,7 @@ bool DecodedVideoProducer::ThreadData::handle_seek()
};
AK::Duration timestamp;
bool moved_position = false;
while (true) {
{
@ -380,8 +407,10 @@ bool DecodedVideoProducer::ThreadData::handle_seek()
}
auto demuxer_seek_result = demuxer_seek_result_or_error.value_or(DemuxerSeekResult::MovedPosition);
if (demuxer_seek_result == DemuxerSeekResult::MovedPosition)
if (demuxer_seek_result == DemuxerSeekResult::MovedPosition) {
m_decoder->flush();
moved_position = true;
}
auto new_seek_id = m_seek_id.load();
RefPtr<VideoFrame> last_frame;
@ -411,7 +440,7 @@ bool DecodedVideoProducer::ThreadData::handle_seek()
if (frame_result.is_error()) {
if (frame_result.error().category() == DecoderErrorCategory::EndOfStream) {
auto locker = take_lock();
resolve_seek(seek_id);
resolve_seek(seek_id, moved_position);
if (last_frame != nullptr)
queue_frame(last_frame.release_nonnull());
return true;
@ -427,8 +456,7 @@ bool DecodedVideoProducer::ThreadData::handle_seek()
auto current_frame = frame_result.release_value();
if (current_frame->timestamp() > timestamp) {
auto locker = take_lock();
m_queue.clear();
resolve_seek(seek_id);
resolve_seek(seek_id, moved_position);
if (last_frame != nullptr)
queue_frame(last_frame.release_nonnull());
@ -463,7 +491,7 @@ void DecodedVideoProducer::ThreadData::push_data_and_decode_some_frames()
while (true) {
{
auto locker = take_lock();
if (m_pending_halting_status == PipelineStatus::Pending)
if (m_current_halting_status == PipelineStatus::Pending)
return;
}
if (handle_seek())

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

@ -14,6 +14,7 @@
#include <AK/Time.h>
#include <LibCore/Forward.h>
#include <LibMedia/DecoderError.h>
#include <LibMedia/Demuxer.h>
#include <LibMedia/Export.h>
#include <LibMedia/Forward.h>
#include <LibMedia/IncrementallyPopulatedStream.h>
@ -49,8 +50,9 @@ public:
void suspend();
void resume();
virtual PipelineStatus pull(RefPtr<VideoFrame>& into) override;
virtual void set_state_changed_handler(PipelineStateChangeHandler) override;
virtual PipelineStatus status() const override;
virtual void pull(RefPtr<VideoFrame>& into) override;
virtual void set_wake_handler(PipelineWakeHandler) override;
AK::Duration select_fast_seek_target(AK::Duration timestamp, SeekMode);
virtual void seek(AK::Duration timestamp) override;
@ -65,7 +67,7 @@ private:
void set_error_handler(ErrorHandler&&);
void set_duration_change_handler(FrameEndTimeHandler&&);
void set_state_changed_handler(PipelineStateChangeHandler);
void set_wake_handler(PipelineWakeHandler);
void start();
DecoderErrorOr<void> create_decoder();
@ -75,7 +77,9 @@ private:
FrameQueue& queue();
PipelineStatus pull(RefPtr<VideoFrame>& into);
PipelineStatus status() const;
PipelineStatus status_while_locked() const;
void pull(RefPtr<VideoFrame>& into);
void seek(AK::Duration timestamp);
AK::Duration select_fast_seek_target(AK::Duration target, SeekMode) const;
@ -92,12 +96,12 @@ private:
void queue_frame(NonnullRefPtr<VideoFrame> const&);
void dispatch_error(DecoderError&&);
bool handle_seek();
void resolve_seek(u32 seek_id);
void resolve_seek(u32 seek_id, bool moved_position);
void push_data_and_decode_some_frames();
void enter_halting_state(PipelineStatus, Optional<DecoderError>);
void dispatch_state_if_changed_while_locked(PipelineStatus);
void dispatch_wake_if_needed_while_locked();
TimeRanges buffered_time_ranges() const;
@ -130,14 +134,15 @@ private:
FrameQueue m_queue;
FrameEndTimeHandler m_duration_change_handler;
ErrorHandler m_error_handler;
PipelineStatus m_pending_halting_status { PipelineStatus::Pending };
PipelineStatus m_current_halting_status { PipelineStatus::Pending };
bool m_moved_position_pending { false };
u32 m_last_processed_seek_id { 0 };
Atomic<u32> m_seek_id { 0 };
AK::Duration m_seek_timestamp;
PipelineStateChangeHandler m_state_changed_handler;
PipelineStatus m_last_dispatched_status { PipelineStatus::Pending };
PipelineWakeHandler m_wake_handler;
mutable bool m_downstream_needs_wake { true };
};
NonnullRefPtr<ThreadData> m_thread_data;

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

@ -20,8 +20,9 @@ public:
virtual void start() = 0;
virtual PipelineStatus pull(RefPtr<VideoFrame>& into) = 0;
virtual void set_state_changed_handler(PipelineStateChangeHandler) = 0;
virtual PipelineStatus status() const = 0;
virtual void pull(RefPtr<VideoFrame>& into) = 0;
virtual void set_wake_handler(PipelineWakeHandler) = 0;
virtual void seek(AK::Duration timestamp) = 0;
};

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

@ -11,6 +11,7 @@
#include <LibCore/Forward.h>
#include <LibMedia/Audio/PlaybackStream.h>
#include <LibMedia/AudioBlock.h>
#include <LibMedia/PipelineStatus.h>
#include <LibMedia/Producers/AudioProducer.h>
#include <LibSync/ConditionVariable.h>
#include <LibSync/Mutex.h>
@ -63,44 +64,63 @@ ErrorOr<NonnullRefPtr<AudioPlaybackSink>> AudioPlaybackSink::try_create(Pipeline
auto thread = TRY(Threading::Thread::try_create("Audio Processor"sv,
[output_thread_data]() -> intptr_t {
while (true) {
size_t tail_index;
u32 seek_id_at_pull;
while (!output_thread_data->m_audio_processor_should_exit) {
RefPtr<AudioProducer> input;
{
Sync::MutexLocker locker { output_thread_data->m_output_mutex };
while (true) {
if (output_thread_data->m_audio_processor_should_exit)
break;
if (output_thread_data->m_seek_id == 0) {
output_thread_data->m_output_condition.wait();
continue;
}
if (output_thread_data->m_input == nullptr) {
output_thread_data->m_output_condition.wait();
continue;
}
if (output_thread_data->m_block_count == OUTPUT_BLOCK_QUEUE_CAPACITY) {
output_thread_data->m_output_condition.wait();
continue;
}
if (output_thread_data->m_waiting_for_upstream_data) {
output_thread_data->m_output_condition.wait();
continue;
}
break;
}
if (output_thread_data->m_audio_processor_should_exit)
return 0;
output_thread_data->m_waiting_for_upstream_data = true;
seek_id_at_pull = output_thread_data->m_seek_id;
tail_index = output_thread_data->m_block_tail;
input = output_thread_data->m_input;
}
auto& output_block = output_thread_data->m_blocks[tail_index];
output_block.clear();
auto status = input->pull(output_block);
auto& output_block = output_thread_data->m_blocks[output_thread_data->m_block_tail];
if (input != nullptr) {
auto status = input->status();
if (status == PipelineStatus::MovedPosition) {
input->pull(output_block);
VERIFY(output_block.is_empty());
Sync::MutexLocker locker { output_thread_data->m_output_mutex };
output_thread_data->m_block_head = 0;
output_thread_data->m_block_tail = 0;
output_thread_data->m_block_count = 0;
output_thread_data->m_last_real_data_end_in_frames = output_thread_data->m_next_frame_to_play;
output_thread_data->m_waiting_for_upstream_data = false;
output_thread_data->m_output_condition.broadcast();
continue;
}
}
u32 seek_id_at_pull;
{
Sync::MutexLocker locker { output_thread_data->m_output_mutex };
if (output_thread_data->m_audio_processor_should_exit)
break;
if (output_thread_data->m_seek_id == 0) {
output_thread_data->m_output_condition.wait();
continue;
}
if (output_thread_data->m_input == nullptr) {
output_thread_data->m_output_condition.wait();
continue;
}
if (output_thread_data->m_block_count == OUTPUT_BLOCK_QUEUE_CAPACITY) {
output_thread_data->m_output_condition.wait();
continue;
}
if (output_thread_data->m_waiting_for_upstream_data) {
output_thread_data->m_output_condition.wait();
continue;
}
if (output_thread_data->m_audio_processor_should_exit)
continue;
output_thread_data->m_waiting_for_upstream_data = true;
seek_id_at_pull = output_thread_data->m_seek_id;
input = output_thread_data->m_input;
}
auto status = input->status();
if (status == PipelineStatus::MovedPosition)
continue;
input->pull(output_block);
{
Sync::MutexLocker locker { output_thread_data->m_output_mutex };
@ -109,7 +129,7 @@ ErrorOr<NonnullRefPtr<AudioPlaybackSink>> AudioPlaybackSink::try_create(Pipeline
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_tail = (output_thread_data->m_block_tail + 1) % OUTPUT_BLOCK_QUEUE_CAPACITY;
output_thread_data->m_block_count++;
if (output_thread_data->m_playback_stream)
@ -119,12 +139,14 @@ ErrorOr<NonnullRefPtr<AudioPlaybackSink>> AudioPlaybackSink::try_create(Pipeline
if (status == PipelineStatus::HaveData)
output_thread_data->m_last_real_data_end_in_frames = output_block.end_timestamp_in_frames();
if (!can_carry_data(output_thread_data->m_last_dispatched_status))
output_thread_data->dispatch_state_if_changed(status, seek_id_at_pull);
}
if (!can_carry_data(output_thread_data->m_last_dispatched_status) && can_carry_data(status))
output_thread_data->dispatch_state_if_changed(status, seek_id_at_pull);
}
}
return 0;
}));
thread->start();
thread->detach();
@ -145,7 +167,7 @@ AudioPlaybackSink::~AudioPlaybackSink()
{
Sync::MutexLocker locker { m_output_thread_data->m_output_mutex };
if (m_output_thread_data->m_input != nullptr)
m_output_thread_data->m_input->set_state_changed_handler(nullptr);
m_output_thread_data->m_input->set_wake_handler(nullptr);
m_output_thread_data->m_input = nullptr;
m_output_thread_data->m_on_state_changed = nullptr;
m_output_thread_data->m_audio_processor_should_exit = true;
@ -154,10 +176,12 @@ AudioPlaybackSink::~AudioPlaybackSink()
ErrorOr<void> AudioPlaybackSink::connect_input(NonnullRefPtr<AudioProducer> const& input)
{
input->set_state_changed_handler([&output_thread_data = *m_output_thread_data](PipelineStatus status) {
input->set_wake_handler([input, &output_thread_data = *m_output_thread_data] {
auto status = input->status();
if (status == PipelineStatus::Pending)
return;
Sync::MutexLocker locker { output_thread_data.m_output_mutex };
output_thread_data.m_last_pull_status = status;
output_thread_data.m_waiting_for_upstream_data = false;
output_thread_data.m_output_condition.broadcast();
});
@ -179,7 +203,7 @@ ErrorOr<void> AudioPlaybackSink::connect_input(NonnullRefPtr<AudioProducer> cons
void AudioPlaybackSink::disconnect_input_while_locked(NonnullRefPtr<AudioProducer> const& input)
{
input->set_state_changed_handler(nullptr);
input->set_wake_handler(nullptr);
m_output_thread_data->m_input = nullptr;
}
@ -380,14 +404,8 @@ void AudioPlaybackSink::seek(AK::Duration time)
Sync::MutexLocker locker { m_output_thread_data->m_output_mutex };
m_output_thread_data->m_seek_id++;
m_output_thread_data->m_block_head = 0;
m_output_thread_data->m_block_tail = 0;
m_output_thread_data->m_block_count = 0;
m_output_thread_data->m_next_frame_to_play = seek_target_in_frames;
m_output_thread_data->m_last_real_data_end_in_frames = seek_target_in_frames;
m_output_thread_data->m_last_pull_status = PipelineStatus::Pending;
m_output_thread_data->m_last_dispatched_status = PipelineStatus::Pending;
@ -408,13 +426,6 @@ void AudioPlaybackSink::seek(AK::Duration time)
self->m_last_stream_time = new_stream_time;
self->m_last_media_time = self->m_temporary_time.release_value();
{
Sync::MutexLocker locker { self->m_output_thread_data->m_output_mutex };
auto pull_status = self->m_output_thread_data->m_last_pull_status;
if (pull_status != PipelineStatus::Pending)
self->m_output_thread_data->dispatch_state_if_changed(pull_status, self->m_output_thread_data->m_seek_id);
}
if (self->m_playing)
self->resume();
});

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

@ -28,7 +28,7 @@ DisplayingVideoSink::DisplayingVideoSink(NonnullRefPtr<MediaTimeProvider> const&
DisplayingVideoSink::~DisplayingVideoSink()
{
if (m_input != nullptr)
m_input->set_state_changed_handler(nullptr);
m_input->set_wake_handler(nullptr);
}
void DisplayingVideoSink::set_time_provider(NonnullRefPtr<MediaTimeProvider> const& provider)
@ -36,15 +36,30 @@ void DisplayingVideoSink::set_time_provider(NonnullRefPtr<MediaTimeProvider> con
m_time_provider = provider;
}
DisplayingVideoSinkUpdateResult DisplayingVideoSink::consume_moved_position_signals(PipelineStatus& status)
{
auto result = DisplayingVideoSinkUpdateResult::NoChange;
while ((status = m_input->status()) == PipelineStatus::MovedPosition) {
m_input->pull(m_next_frame);
VERIFY(m_next_frame == nullptr);
m_current_frame.clear();
result = DisplayingVideoSinkUpdateResult::NewFrameAvailable;
}
VERIFY(status != PipelineStatus::MovedPosition);
return result;
}
ErrorOr<void> DisplayingVideoSink::connect_input(NonnullRefPtr<VideoProducer> const& input)
{
VERIFY(m_input == nullptr);
m_input = input;
input->set_state_changed_handler([this](PipelineStatus status) {
if (!m_seek_pending_display_update)
return;
input->set_wake_handler([this, input] {
auto status = PipelineStatus::Pending;
if (consume_moved_position_signals(status) == DisplayingVideoSinkUpdateResult::NewFrameAvailable)
m_seek_status = SeekStatus::FrameInvalidated;
if (is_waiting_for_data(status))
return;
m_seek_status = SeekStatus::Complete;
dispatch_state_if_changed(status);
});
input->seek(m_time_provider->current_time());
@ -55,7 +70,7 @@ ErrorOr<void> DisplayingVideoSink::connect_input(NonnullRefPtr<VideoProducer> co
void DisplayingVideoSink::disconnect_input(NonnullRefPtr<VideoProducer> const& input)
{
VERIFY(m_input == input);
m_input->set_state_changed_handler(nullptr);
m_input->set_wake_handler(nullptr);
m_input = nullptr;
}
@ -63,9 +78,8 @@ void DisplayingVideoSink::seek(AK::Duration timestamp)
{
if (m_input != nullptr)
m_input->seek(timestamp);
m_next_frame.clear();
m_last_dispatched_status = PipelineStatus::Pending;
m_seek_pending_display_update = true;
m_seek_status = SeekStatus::InProgress;
}
void DisplayingVideoSink::dispatch_state_if_changed(PipelineStatus status)
@ -85,33 +99,33 @@ DisplayingVideoSinkUpdateResult DisplayingVideoSink::update()
auto current_time = m_time_provider->current_time();
auto result = DisplayingVideoSinkUpdateResult::NoChange;
auto last_pull_status = PipelineStatus::Pending;
if (m_seek_status == SeekStatus::FrameInvalidated) {
result = DisplayingVideoSinkUpdateResult::NewFrameAvailable;
m_seek_status = SeekStatus::None;
} else if (m_seek_status == SeekStatus::Complete) {
result = DisplayingVideoSinkUpdateResult::NewFrameAvailable;
}
auto last_status = PipelineStatus::Pending;
while (true) {
if (consume_moved_position_signals(last_status) == DisplayingVideoSinkUpdateResult::NewFrameAvailable)
result = DisplayingVideoSinkUpdateResult::NewFrameAvailable;
if (m_next_frame == nullptr) {
last_pull_status = m_input->pull(m_next_frame);
if (last_pull_status != PipelineStatus::HaveData)
if (last_status != PipelineStatus::HaveData)
break;
m_input->pull(m_next_frame);
VERIFY(m_next_frame != nullptr);
m_seek_status = SeekStatus::Complete;
}
if (m_seek_status != SeekStatus::Complete)
break;
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;
if (m_seek_pending_display_update && !is_waiting_for_data(effective_status)) {
if (result != DisplayingVideoSinkUpdateResult::NewFrameAvailable) {
m_current_frame.clear();
result = DisplayingVideoSinkUpdateResult::NewFrameAvailable;
}
m_seek_pending_display_update = false;
}
dispatch_state_if_changed(effective_status);
dispatch_state_if_changed(last_status);
return result;
}

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

@ -17,7 +17,7 @@
namespace Media {
enum class DisplayingVideoSinkUpdateResult : u8 {
enum class [[nodiscard]] DisplayingVideoSinkUpdateResult : u8 {
NewFrameAvailable,
NoChange,
};
@ -40,6 +40,8 @@ public:
RefPtr<VideoFrame> current_frame();
private:
DisplayingVideoSinkUpdateResult consume_moved_position_signals(PipelineStatus&);
void dispatch_state_if_changed(PipelineStatus);
NonnullRefPtr<MediaTimeProvider> m_time_provider;
@ -47,7 +49,14 @@ private:
RefPtr<VideoFrame> m_next_frame;
RefPtr<VideoFrame> m_current_frame;
bool m_seek_pending_display_update { false };
enum class SeekStatus : u8 {
None,
InProgress,
FrameInvalidated,
Complete,
};
SeekStatus m_seek_status { SeekStatus::None };
PipelineStateChangeHandler m_on_state_changed;
PipelineStatus m_last_dispatched_status { PipelineStatus::Pending };

4
Tests/LibMedia/TestDataProducers.cpp
View file

@ -127,7 +127,9 @@ TEST_CASE(audio_producer_underspecified_5_1_channel_map)
while (true) {
Media::AudioBlock block;
auto status = producer->pull(block);
auto status = producer->status();
if (status == Media::PipelineStatus::HaveData)
producer->pull(block);
if (status == Media::PipelineStatus::HaveData) {
EXPECT(!block.is_empty());
EXPECT_EQ(block.channel_count(), 6);

4
Tests/LibMedia/TestFFmpegAudioNormalization.cpp
View file

@ -94,7 +94,9 @@ static void decode_and_expect()
MonotonicTime deadline = MonotonicTime::now_coarse() + AK::Duration::from_seconds(1);
while (MonotonicTime::now_coarse() < deadline) {
Media::AudioBlock block;
auto status = producer->pull(block);
auto status = producer->status();
if (status == Media::PipelineStatus::HaveData)
producer->pull(block);
if (status == Media::PipelineStatus::HaveData) {
EXPECT(!block.is_empty());
for (float sample : block.data()) {

4
Tests/LibMedia/TestMediaCommon.h
View file

@ -99,7 +99,9 @@ static inline void decode_audio(StringView path, u32 sample_rate, u8 channel_cou
while (true) {
Media::AudioBlock block;
auto status = producer->pull(block);
auto status = producer->status();
if (status == Media::PipelineStatus::HaveData)
producer->pull(block);
if (status == Media::PipelineStatus::HaveData) {
EXPECT(!block.is_empty());
EXPECT_EQ(block.sample_rate(), sample_rate);