From f62af69dcd3521378d2b22da4f8cb68917579a03 Mon Sep 17 00:00:00 2001 From: Zaggy1024 Date: Sat, 28 Mar 2026 20:43:00 -0500 Subject: [PATCH] LibMedia: Allow MediaStreamCursor to never block on missing data This is necessary to allow the main thread to read out frames to determine the buffered ranges, without affecting the request position of an IncrementallyPopulatedStream or causing a deadlock waiting for data to come in. --- .../LibMedia/IncrementallyPopulatedStream.cpp | 56 +++++++++++-------- .../LibMedia/IncrementallyPopulatedStream.h | 5 +- Libraries/LibMedia/MediaStream.h | 2 + Libraries/LibMedia/ReadonlyBytesCursor.h | 2 + 4 files changed, 42 insertions(+), 23 deletions(-) diff --git a/Libraries/LibMedia/IncrementallyPopulatedStream.cpp b/Libraries/LibMedia/IncrementallyPopulatedStream.cpp index 824cdd3e2e..008802cbd3 100644 --- a/Libraries/LibMedia/IncrementallyPopulatedStream.cpp +++ b/Libraries/LibMedia/IncrementallyPopulatedStream.cpp @@ -162,7 +162,7 @@ static u64 adjust_request_position(u64 position) return 0; } -bool IncrementallyPopulatedStream::check_if_data_is_available_or_begin_request_while_locked(MonotonicTime now, u64 position, u64 length) +bool IncrementallyPopulatedStream::check_if_data_is_available_or_begin_request_while_locked(Cursor& cursor, u64 position, u64 length) { auto* chunk = m_chunks.find_largest_not_above(position); if (!chunk) @@ -170,23 +170,30 @@ bool IncrementallyPopulatedStream::check_if_data_is_available_or_begin_request_w VERIFY(position >= chunk->offset()); - auto potential_request_position = adjust_request_position(position); - potential_request_position = max(chunk->end(), position); - for (size_t i = 0; i < m_cursors.size(); i++) { - auto const& other_cursor = m_cursors[i]; - if (now >= other_cursor.m_active_timeout && !other_cursor.m_blocked) - continue; - if (other_cursor.m_position < potential_request_position) { - auto* other_cursor_chunk = m_chunks.find_largest_not_above(other_cursor.m_position); - if (other_cursor_chunk && other_cursor_chunk->end() >= other_cursor.m_position) { - potential_request_position = other_cursor_chunk->end(); + if (cursor.m_is_blocking) { + auto now = MonotonicTime::now_coarse(); + cursor.m_active_timeout = now + CURSOR_ACTIVE_TIME; + + auto potential_request_position = adjust_request_position(position); + potential_request_position = max(chunk->end(), position); + for (size_t i = 0; i < m_cursors.size(); i++) { + auto const& other_cursor = m_cursors[i]; + if (!other_cursor.m_is_blocking) continue; + if (now >= other_cursor.m_active_timeout && !other_cursor.m_blocked) + continue; + if (other_cursor.m_position < potential_request_position) { + auto* other_cursor_chunk = m_chunks.find_largest_not_above(other_cursor.m_position); + if (other_cursor_chunk && other_cursor_chunk->end() >= other_cursor.m_position) { + potential_request_position = other_cursor_chunk->end(); + continue; + } + potential_request_position = other_cursor.m_position; } - potential_request_position = other_cursor.m_position; } + if (m_currently_requested_position > potential_request_position || potential_request_position > m_last_chunk_end + FORWARD_REQUEST_THRESHOLD) + begin_new_request_while_locked(potential_request_position); } - if (m_currently_requested_position > potential_request_position || potential_request_position > m_last_chunk_end + FORWARD_REQUEST_THRESHOLD) - begin_new_request_while_locked(potential_request_position); u64 end = position + length; if (m_closed && end > m_expected_size.value()) @@ -197,14 +204,17 @@ bool IncrementallyPopulatedStream::check_if_data_is_available_or_begin_request_w size_t IncrementallyPopulatedStream::read_from_chunks_while_locked(u64 position, Bytes& bytes) const { auto chunk_iterator = m_chunks.find_largest_not_above_iterator(position); - VERIFY(!chunk_iterator.is_end()); + if (chunk_iterator.is_end()) + return 0; + auto const& chunk = *chunk_iterator; + if (position >= chunk.end()) + return 0; + VERIFY(position >= chunk.offset()); auto end = position + bytes.size(); auto copy_size = bytes.size(); if (end > chunk.end()) { - VERIFY(m_expected_size.has_value()); - VERIFY(chunk.end() == m_expected_size.value()); end = chunk.end(); copy_size = end - position; } @@ -219,11 +229,8 @@ DecoderErrorOr IncrementallyPopulatedStream::read_at(Cursor& cursor, siz { Sync::MutexLocker locker { m_mutex }; - auto now = MonotonicTime::now_coarse(); - cursor.m_active_timeout = now + CURSOR_ACTIVE_TIME; - - while (!cursor.m_aborted) { - if (check_if_data_is_available_or_begin_request_while_locked(now, position, bytes.size())) + while (!cursor.m_aborted && cursor.m_is_blocking) { + if (check_if_data_is_available_or_begin_request_while_locked(cursor, position, bytes.size())) break; cursor.m_blocked = true; @@ -261,6 +268,11 @@ IncrementallyPopulatedStream::Cursor::~Cursor() VERIFY(m_stream->m_cursors.remove_first_matching([&](Cursor const& cursor) { return this == &cursor; })); } +void IncrementallyPopulatedStream::Cursor::set_is_blocking(bool blocking) +{ + m_is_blocking = blocking; +} + DecoderErrorOr IncrementallyPopulatedStream::Cursor::seek(i64 offset, AK::SeekMode mode) { switch (mode) { diff --git a/Libraries/LibMedia/IncrementallyPopulatedStream.h b/Libraries/LibMedia/IncrementallyPopulatedStream.h index 93bc3dcc23..e1b625ba38 100644 --- a/Libraries/LibMedia/IncrementallyPopulatedStream.h +++ b/Libraries/LibMedia/IncrementallyPopulatedStream.h @@ -51,6 +51,8 @@ public: public: ~Cursor(); + virtual void set_is_blocking(bool) override; + virtual DecoderErrorOr seek(i64 offset, AK::SeekMode mode) override; virtual DecoderErrorOr read_into(Bytes bytes) override; @@ -69,6 +71,7 @@ public: Cursor(NonnullRefPtr const& stream); NonnullRefPtr m_stream; + bool m_is_blocking { true }; size_t m_position { 0 }; bool m_aborted { false }; Atomic m_blocked { false }; @@ -106,7 +109,7 @@ private: DecoderErrorOr read_at(Cursor&, size_t position, Bytes&); void begin_new_request_while_locked(u64 position); - bool check_if_data_is_available_or_begin_request_while_locked(MonotonicTime now, u64 position, u64 length); + bool check_if_data_is_available_or_begin_request_while_locked(Cursor&, u64 position, u64 length); size_t read_from_chunks_while_locked(u64 position, Bytes& bytes) const; mutable Sync::Mutex m_mutex; diff --git a/Libraries/LibMedia/MediaStream.h b/Libraries/LibMedia/MediaStream.h index 8b5a82536a..4eee59810a 100644 --- a/Libraries/LibMedia/MediaStream.h +++ b/Libraries/LibMedia/MediaStream.h @@ -17,6 +17,8 @@ class MediaStreamCursor : public AtomicRefCounted { public: virtual ~MediaStreamCursor() = default; + virtual void set_is_blocking(bool) = 0; + virtual DecoderErrorOr seek(i64 offset, AK::SeekMode) = 0; virtual DecoderErrorOr read_into(Bytes) = 0; virtual size_t position() const = 0; diff --git a/Libraries/LibMedia/ReadonlyBytesCursor.h b/Libraries/LibMedia/ReadonlyBytesCursor.h index f6e48bfa62..7c5d96a74e 100644 --- a/Libraries/LibMedia/ReadonlyBytesCursor.h +++ b/Libraries/LibMedia/ReadonlyBytesCursor.h @@ -19,6 +19,8 @@ public: { } + virtual void set_is_blocking(bool) override { } + virtual DecoderErrorOr seek(i64 offset, AK::SeekMode mode) override { auto target_position = [&] -> size_t {