diff --git a/Libraries/LibMedia/AudioBlock.h b/Libraries/LibMedia/AudioBlock.h index 92425db1e8..30753b788e 100644 --- a/Libraries/LibMedia/AudioBlock.h +++ b/Libraries/LibMedia/AudioBlock.h @@ -41,6 +41,15 @@ public: m_timestamp_in_samples = timestamp.to_time_units(1, sample_rate()); data_callback(m_data); } + template + void emplace(Audio::SampleSpecification sample_specification, i64 timestamp_in_samples, Callback data_callback) + { + VERIFY(sample_specification.is_valid()); + m_sample_specification = sample_specification; + m_timestamp_in_samples = timestamp_in_samples; + m_timestamp = AK::Duration::from_time_units(timestamp_in_samples, 1, sample_rate()); + data_callback(m_data); + } u32 sample_rate() const { return sample_specification().sample_rate(); diff --git a/Libraries/LibMedia/Sinks/AudioMixingSink.cpp b/Libraries/LibMedia/Sinks/AudioMixingSink.cpp index 6989a08c1c..0b5dcbb72f 100644 --- a/Libraries/LibMedia/Sinks/AudioMixingSink.cpp +++ b/Libraries/LibMedia/Sinks/AudioMixingSink.cpp @@ -4,25 +4,134 @@ * SPDX-License-Identifier: BSD-2-Clause */ +#include +#include +#include +#include #include #include +#include #include +#include +#include #include "AudioMixingSink.h" namespace Media { +static constexpr size_t MAX_SAMPLES_PER_OUTPUT_BLOCK = 1024; +static constexpr size_t OUTPUT_BLOCK_QUEUE_CAPACITY = 4; + +class AudioMixingSink::OutputThreadData : public AtomicRefCounted { +public: + struct TrackMixingData { + TrackMixingData(NonnullRefPtr const& provider) + : provider(provider) + { + } + + NonnullRefPtr provider; + AudioBlock current_block; + bool buffering { false }; + }; + + OutputThreadData() = default; + + bool mix_one_block_into(AudioBlock&); + ReadonlySpan move_output_to_playback_stream_buffer(Span); + + RefPtr m_playback_stream; + + mutable Sync::Mutex m_output_mutex; + mutable Sync::ConditionVariable m_output_condition { m_output_mutex }; + + AK::Array m_blocks; + size_t m_block_head { 0 }; + size_t m_block_tail { 0 }; + size_t m_block_count { 0 }; + i64 m_next_sample_to_play { 0 }; + + Atomic m_pause_writing_audio_data { true }; + bool m_filler_is_waiting_in_output_loop { false }; + bool m_filler_should_exit { false }; + + mutable Sync::Mutex m_mixing_data_mutex; + Audio::SampleSpecification m_sample_specification; + HashMap m_track_mixing_datas; + Atomic m_next_sample_to_write { 0 }; + + Function on_track_started_buffering; +}; + ErrorOr> AudioMixingSink::try_create() { auto weak_ref = TRY(try_make_ref_counted()); - auto sink = TRY(try_make_ref_counted(weak_ref)); + auto output_thread_data = TRY(adopt_nonnull_ref_or_enomem(new (nothrow) OutputThreadData)); + auto sink = TRY(try_make_ref_counted(weak_ref, output_thread_data)); weak_ref->emplace(sink); + + output_thread_data->on_track_started_buffering = [weak_self = sink->m_weak_self, &event_loop = sink->m_main_thread_event_loop](Track const& track) { + event_loop.deferred_invoke([weak_self, track] { + auto self = weak_self->take_strong(); + if (self && self->on_start_buffering) + self->on_start_buffering(track); + }); + }; + + auto thread = TRY(Threading::Thread::try_create("Audio Mixer"sv, + [output_thread_data]() -> intptr_t { + while (true) { + 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_output_condition.broadcast(); + while (true) { + if (output_thread_data->m_filler_should_exit) + break; + if (output_thread_data->m_pause_writing_audio_data) { + 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; + } + break; + } + if (output_thread_data->m_filler_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; + tail_index = output_thread_data->m_block_tail; + } + + if (!output_thread_data->mix_one_block_into(output_thread_data->m_blocks[tail_index])) + continue; + + { + 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++; + + if (output_thread_data->m_playback_stream) + output_thread_data->m_playback_stream->notify_data_available(); + + output_thread_data->m_output_condition.broadcast(); + } + } + })); + thread->start(); + thread->detach(); + return sink; } -AudioMixingSink::AudioMixingSink(AudioMixingSinkWeakReference& weak_ref) +AudioMixingSink::AudioMixingSink(AudioMixingSinkWeakReference& weak_ref, NonnullRefPtr output_thread_data) : m_main_thread_event_loop(Core::EventLoop::current()) , m_weak_self(weak_ref) + , m_output_thread_data(move(output_thread_data)) { m_main_thread_event_loop.deferred_invoke([weak_self = m_weak_self] { auto self = weak_self->take_strong(); @@ -34,28 +143,36 @@ AudioMixingSink::AudioMixingSink(AudioMixingSinkWeakReference& weak_ref) AudioMixingSink::~AudioMixingSink() { + { + Sync::MutexLocker locker { m_output_thread_data->m_output_mutex }; + m_output_thread_data->m_filler_should_exit = true; + m_output_thread_data->m_output_condition.broadcast(); + } m_weak_self->revoke(); } void AudioMixingSink::set_provider(Track const& track, RefPtr const& provider) { - Sync::MutexLocker locker { m_mutex }; - m_track_mixing_datas.remove(track); - if (provider == nullptr) - return; + { + Sync::MutexLocker locker { m_output_thread_data->m_mixing_data_mutex }; + m_output_thread_data->m_track_mixing_datas.remove(track); + if (provider == nullptr) + return; - // The provider must have its output sample specification set before it starts decoding, or - // we'll drop some samples due to a mismatch. - m_track_mixing_datas.set(track, TrackMixingData(*provider)); - if (m_sample_specification.is_valid()) { - provider->set_output_sample_specification(m_sample_specification); + // The provider must have its output sample specification set before it starts decoding, or + // we'll drop some samples due to a mismatch. + m_output_thread_data->m_track_mixing_datas.set(track, OutputThreadData::TrackMixingData(*provider)); + } + + if (m_output_thread_data->m_sample_specification.is_valid()) { + provider->set_output_sample_specification(m_output_thread_data->m_sample_specification); provider->start(); } } RefPtr AudioMixingSink::provider(Track const& track) const { - auto mixing_data = m_track_mixing_datas.get(track); + auto mixing_data = m_output_thread_data->m_track_mixing_datas.get(track); if (!mixing_data.has_value()) return nullptr; return mixing_data->provider; @@ -68,11 +185,8 @@ void AudioMixingSink::create_playback_stream() m_started_creating_playback_stream = true; - auto data_callback = [weak_self = m_weak_self](Span buffer) -> ReadonlySpan { - auto self = weak_self->take_strong(); - if (!self) - return buffer.trim(0); - return self->write_audio_data_to_playback_stream(buffer); + auto data_callback = [output_thread_data = m_output_thread_data](Span buffer) -> ReadonlySpan { + return output_thread_data->move_output_to_playback_stream_buffer(buffer); }; constexpr u32 target_latency_ms = 100; @@ -83,19 +197,26 @@ void AudioMixingSink::create_playback_stream() if (!self) return; - self->m_playback_stream = stream; + self->m_output_thread_data->m_playback_stream = stream; + self->m_output_thread_data->m_sample_specification = stream->sample_specification(); self->set_volume(self->m_volume); - if (self->m_temporary_time.has_value()) - self->set_time(self->m_temporary_time.release_value()); - Sync::MutexLocker locker { self->m_mutex }; - self->m_sample_specification = stream->sample_specification(); - - for (auto& [track, track_data] : self->m_track_mixing_datas) { - track_data.provider->set_output_sample_specification(self->m_sample_specification); + for (auto& [track, track_data] : self->m_output_thread_data->m_track_mixing_datas) { + track_data.provider->set_output_sample_specification(self->m_output_thread_data->m_sample_specification); track_data.provider->start(); } + if (self->m_temporary_time.has_value()) { + self->set_time(self->m_temporary_time.release_value()); + return; + } + + { + Sync::MutexLocker locker { self->m_output_thread_data->m_output_mutex }; + self->m_output_thread_data->m_pause_writing_audio_data = false; + self->m_output_thread_data->m_output_condition.broadcast(); + } + if (self->m_playing) self->resume(); }); @@ -110,30 +231,50 @@ void AudioMixingSink::create_playback_stream() }); } -ReadonlySpan AudioMixingSink::write_audio_data_to_playback_stream(Span buffer) +bool AudioMixingSink::OutputThreadData::mix_one_block_into(AudioBlock& out_block) { VERIFY(m_sample_specification.is_valid()); - VERIFY(buffer.size() > 0); auto channel_count = m_sample_specification.channel_count(); - auto sample_count = buffer.size() / channel_count; - buffer.fill(0.0f); + auto max_sample_count = MAX_SAMPLES_PER_OUTPUT_BLOCK / channel_count; - Sync::MutexLocker mixing_data_locker { m_mutex }; + Sync::MutexLocker mixing_data_locker { m_mixing_data_mutex }; auto buffer_start = m_next_sample_to_write.load(); - auto samples_end = buffer_start + static_cast(sample_count); + auto initial_samples_end = buffer_start + static_cast(max_sample_count); + auto samples_end = initial_samples_end; auto buffering = false; + auto any_track_has_fresh_data = false; for (auto& [track, track_data] : m_track_mixing_datas) { - if (!track_data.provider->is_blocked()) - continue; auto available_end = track_data.provider->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.provider->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; - buffering = true; + if (track_data.provider->is_blocked()) + buffering = true; } } + if (!m_track_mixing_datas.is_empty() && !any_track_has_fresh_data) + return false; + for (auto& [track, track_data] : m_track_mixing_datas) { if (!buffering) { track_data.buffering = false; @@ -144,111 +285,163 @@ ReadonlySpan AudioMixingSink::write_audio_data_to_playback_stream(Spantake_strong(); - if (self && self->on_start_buffering) - self->on_start_buffering(track); - }); + if (on_track_started_buffering) + on_track_started_buffering(track); } } - sample_count = max(samples_end - buffer_start, 0); + auto sample_count = static_cast(max(samples_end - buffer_start, 0)); auto write_size = sample_count * channel_count; if (sample_count == 0) - return buffer; + return false; - for (auto& [track, track_data] : m_track_mixing_datas) { - auto next_sample = buffer_start; + out_block.emplace(m_sample_specification, buffer_start, [&](AudioBlock::Data& data) { + if (data.size() != write_size) + data = MUST(AudioBlock::Data::create(write_size)); + for (size_t i = 0; i < write_size; i++) + data[i] = 0.0f; - auto go_to_next_block = [&] { - auto new_block = track_data.provider->retrieve_block(); - if (new_block.is_empty()) - return false; + for (auto& [track, track_data] : m_track_mixing_datas) { + auto next_sample = buffer_start; - track_data.current_block = move(new_block); - return true; - }; + auto go_to_next_block = [&] { + auto new_block = track_data.provider->retrieve_block(); + if (new_block.is_empty()) + return false; - if (track_data.current_block.is_empty()) { - if (!go_to_next_block()) - continue; + track_data.current_block = move(new_block); + return true; + }; + + if (track_data.current_block.is_empty()) { + if (!go_to_next_block()) + continue; + } + + while (!track_data.current_block.is_empty()) { + auto& current_block = track_data.current_block; + auto current_block_sample_count = static_cast(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((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((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(write_count / channel_count); + if (next_sample == samples_end) + break; + VERIFY(next_sample < samples_end); + } + } + }); + + m_next_sample_to_write += static_cast(sample_count); + return true; +} + +ReadonlySpan AudioMixingSink::OutputThreadData::move_output_to_playback_stream_buffer(Span buffer) +{ + VERIFY(buffer.size() > 0); + + Sync::MutexLocker locker { m_output_mutex }; + + size_t samples_written = 0; + while (samples_written < buffer.size() && m_block_count > 0) { + auto const& head_block = m_blocks[m_block_head]; + auto channel_count = head_block.channel_count(); + auto block_start_sample = head_block.timestamp_in_samples(); + auto block_end_sample = block_start_sample + static_cast(head_block.sample_count()); + + if (m_next_sample_to_play >= block_end_sample) { + m_block_head = (m_block_head + 1) % OUTPUT_BLOCK_QUEUE_CAPACITY; + m_block_count--; + continue; } - while (!track_data.current_block.is_empty()) { - auto& current_block = track_data.current_block; - auto current_block_sample_count = static_cast(current_block.sample_count()); + if (m_next_sample_to_play < block_start_sample) { + auto silence_samples = static_cast(block_start_sample - m_next_sample_to_play) * channel_count; + auto samples_to_silence = min(silence_samples, buffer.size() - samples_written); + for (size_t i = 0; i < samples_to_silence; i++) + buffer[samples_written + i] = 0.0f; + samples_written += samples_to_silence; + m_next_sample_to_play += static_cast(samples_to_silence / channel_count); + continue; + } - if (current_block.sample_specification() != m_sample_specification) { - if (!go_to_next_block()) - break; - current_block.clear(); - continue; - } + auto offset_in_head_samples = static_cast(m_next_sample_to_play - block_start_sample) * channel_count; + auto samples_remaining_in_head = head_block.data_count() - offset_in_head_samples; + auto samples_to_copy = min(samples_remaining_in_head, buffer.size() - samples_written); - auto first_sample_offset = current_block.timestamp_in_samples(); - if (first_sample_offset >= samples_end) - break; + for (size_t i = 0; i < samples_to_copy; i++) + buffer[samples_written + i] = head_block.data()[offset_in_head_samples + i]; - auto block_end = first_sample_offset + current_block_sample_count; - if (block_end <= next_sample) { - if (!go_to_next_block()) - break; - continue; - } + samples_written += samples_to_copy; + m_next_sample_to_play += static_cast(samples_to_copy / channel_count); - next_sample = max(next_sample, first_sample_offset); - - VERIFY(next_sample >= first_sample_offset); - auto index_in_block = static_cast((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((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++) - buffer[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(write_count / channel_count); - if (next_sample == samples_end) - break; - VERIFY(next_sample < samples_end); + if (offset_in_head_samples + samples_to_copy == head_block.data_count()) { + m_block_head = (m_block_head + 1) % OUTPUT_BLOCK_QUEUE_CAPACITY; + m_block_count--; } } - m_next_sample_to_write += static_cast(sample_count); + if (samples_written < buffer.size()) + buffer = buffer.trim(samples_written); + + m_output_condition.broadcast(); return buffer; } AK::Duration AudioMixingSink::current_time() const { - if (!m_sample_specification.is_valid()) - return AK::Duration::zero(); if (m_temporary_time.has_value()) return m_temporary_time.value(); - if (!m_playback_stream) + if (!m_output_thread_data->m_playback_stream) return m_last_media_time; - auto time = m_last_media_time + (m_playback_stream->total_time_played() - m_last_stream_time); - auto max_time = AK::Duration::from_time_units(m_next_sample_to_write.load(MemoryOrder::memory_order_acquire), 1, m_sample_specification.sample_rate()); - time = min(time, max_time); - return time; + auto stream_time = m_output_thread_data->m_playback_stream->total_time_played(); + return m_last_media_time + (stream_time - m_last_stream_time); } void AudioMixingSink::resume() @@ -259,9 +452,9 @@ void AudioMixingSink::resume() if (m_temporary_time.has_value()) return; - if (!m_playback_stream) + if (!m_output_thread_data->m_playback_stream) return; - m_playback_stream->resume() + m_output_thread_data->m_playback_stream->resume() ->when_resolved([weak_self = m_weak_self](auto new_stream_time) { auto self = weak_self->take_strong(); if (!self) @@ -282,9 +475,9 @@ void AudioMixingSink::pause() { m_playing = false; - if (!m_playback_stream) + if (!m_output_thread_data->m_playback_stream) return; - m_playback_stream->drain_buffer_and_suspend() + m_output_thread_data->m_playback_stream->drain_buffer_and_suspend() ->when_resolved([]() { }) .when_rejected([](auto&& error) { @@ -304,28 +497,42 @@ void AudioMixingSink::set_time(AK::Duration time) m_temporary_time = time; - if (!m_playback_stream) + if (!m_output_thread_data->m_playback_stream) return; - m_playback_stream->drain_buffer_and_suspend() + m_output_thread_data->m_pause_writing_audio_data = true; + + m_output_thread_data->m_playback_stream->drain_buffer_and_suspend() ->when_resolved([weak_self = m_weak_self]() { auto self = weak_self->take_strong(); if (!self) return; - auto new_stream_time = self->m_playback_stream->total_time_played(); + auto new_stream_time = self->m_output_thread_data->m_playback_stream->total_time_played(); self->m_main_thread_event_loop.deferred_invoke([self, new_stream_time]() { self->m_last_stream_time = new_stream_time; self->m_last_media_time = self->m_temporary_time.release_value(); - { - Sync::MutexLocker mixing_locker { self->m_mutex }; - self->m_next_sample_to_write = self->m_last_media_time.to_time_units(1, self->m_sample_specification.sample_rate()); - } + auto seek_target_in_samples = self->m_last_media_time.to_time_units(1, self->m_output_thread_data->m_sample_specification.sample_rate()); - for (auto& [track, track_data] : self->m_track_mixing_datas) - track_data.current_block.clear(); + { + 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) + 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_next_sample_to_write = seek_target_in_samples; + for (auto& [track, track_data] : self->m_output_thread_data->m_track_mixing_datas) + track_data.current_block.clear(); + + self->m_output_thread_data->m_next_sample_to_play = seek_target_in_samples; + + self->m_output_thread_data->m_pause_writing_audio_data = false; + self->m_output_thread_data->m_output_condition.broadcast(); + } if (self->m_playing) self->resume(); @@ -340,8 +547,8 @@ void AudioMixingSink::set_volume(double volume) { m_volume = volume; - if (m_playback_stream) { - m_playback_stream->set_volume(m_volume) + if (m_output_thread_data->m_playback_stream) { + m_output_thread_data->m_playback_stream->set_volume(m_volume) ->when_rejected([](Error&&) { // FIXME: Do we even need this function to return a promise? }); diff --git a/Libraries/LibMedia/Sinks/AudioMixingSink.h b/Libraries/LibMedia/Sinks/AudioMixingSink.h index 208d036a00..810279373f 100644 --- a/Libraries/LibMedia/Sinks/AudioMixingSink.h +++ b/Libraries/LibMedia/Sinks/AudioMixingSink.h @@ -6,7 +6,6 @@ #pragma once -#include #include #include #include @@ -15,7 +14,6 @@ #include #include #include -#include #include namespace Media { @@ -23,9 +21,12 @@ namespace Media { class MEDIA_API AudioMixingSink final : public AudioSink { class AudioMixingSinkWeakReference; +private: + class OutputThreadData; + public: static ErrorOr> try_create(); - AudioMixingSink(AudioMixingSinkWeakReference&); + AudioMixingSink(AudioMixingSinkWeakReference&, NonnullRefPtr); virtual ~AudioMixingSink() override; virtual void set_provider(Track const&, RefPtr const&) override; @@ -45,8 +46,6 @@ public: Function on_start_buffering; private: - static constexpr size_t MAX_BLOCK_COUNT = 16; - class AudioMixingSinkWeakReference : public AtomicRefCounted { public: void emplace(AudioMixingSink& sink) { m_ptr = &sink; } @@ -66,37 +65,20 @@ private: AudioMixingSink* m_ptr { nullptr }; }; - struct TrackMixingData { - TrackMixingData(NonnullRefPtr const& provider) - : provider(provider) - { - } - - NonnullRefPtr provider; - AudioBlock current_block; - bool buffering { false }; - }; - void create_playback_stream(); - ReadonlySpan write_audio_data_to_playback_stream(Span); Core::EventLoop& m_main_thread_event_loop; NonnullRefPtr m_weak_self; - Sync::Mutex m_mutex; - Sync::ConditionVariable m_wait_condition { m_mutex }; bool m_started_creating_playback_stream { false }; - RefPtr m_playback_stream; - Audio::SampleSpecification m_sample_specification; bool m_playing { false }; double m_volume { 1 }; - HashMap m_track_mixing_datas; - Atomic m_next_sample_to_write { 0 }; - AK::Duration m_last_stream_time; AK::Duration m_last_media_time; Optional m_temporary_time; + + NonnullRefPtr m_output_thread_data; }; }