From 88dfee5528eda910d0e0412314f39b9fd77fa090 Mon Sep 17 00:00:00 2001 From: Andreas Kling Date: Fri, 5 Jun 2026 00:47:10 +0200 Subject: [PATCH] ImageDecoder: Decode images on ThreadPool Dispatch image decode work to ThreadPool workers instead of the old BackgroundAction helper. BackgroundAction funneled all jobs through one background thread, so image decodes were still serialized. The pool lets independent decode jobs run in parallel while completions are posted back to the origin event loop. Keep pending decode state as a small cancel flag. Make animated sessions ref-counted with a mutex so overlapping frame requests cannot race shared decoders once jobs run in parallel. --- .../ImageDecoder/ConnectionFromClient.cpp | 162 ++++++++++++------ Services/ImageDecoder/ConnectionFromClient.h | 26 ++- 2 files changed, 126 insertions(+), 62 deletions(-) diff --git a/Services/ImageDecoder/ConnectionFromClient.cpp b/Services/ImageDecoder/ConnectionFromClient.cpp index 2c275a81ba..b10eaa3a12 100644 --- a/Services/ImageDecoder/ConnectionFromClient.cpp +++ b/Services/ImageDecoder/ConnectionFromClient.cpp @@ -6,14 +6,18 @@ #include #include +#include #include #include +#include #include #include #include #include #include #include +#include +#include namespace ImageDecoder { @@ -42,7 +46,6 @@ void ConnectionFromClient::die() s_client_ids.deallocate(client_id); if (s_connections.is_empty()) { - Threading::quit_background_thread(); Core::Process::terminate_immediately(0); } } @@ -176,33 +179,50 @@ static ErrorOr decode_image_to_details(Core: return result; } -NonnullRefPtr ConnectionFromClient::make_decode_image_job(i64 request_id, Core::AnonymousBuffer encoded_buffer, Optional ideal_size, Optional mime_type) +NonnullRefPtr ConnectionFromClient::start_decode_image_job(i64 request_id, Core::AnonymousBuffer encoded_buffer, Optional ideal_size, Optional mime_type) { - return Job::construct( - [encoded_buffer = move(encoded_buffer), ideal_size = move(ideal_size), mime_type = move(mime_type)](auto&) mutable -> ErrorOr { - return TRY(decode_image_to_details(move(encoded_buffer), ideal_size, mime_type)); - }, - [strong_this = NonnullRefPtr(*this), request_id](DecodeResult result) { - i64 session_id = 0; + auto job = make_ref_counted(); + auto& main_thread_event_loop = Core::EventLoop::current(); + Threading::ThreadPool::the().submit( + [strong_this = NonnullRefPtr(*this), job, &main_thread_event_loop, request_id, encoded_buffer = move(encoded_buffer), ideal_size = move(ideal_size), mime_type = move(mime_type)]() mutable { + auto result = decode_image_to_details(move(encoded_buffer), ideal_size, mime_type); - if (result.decoder) { - // This is a streaming animated decode. Create a session. - session_id = strong_this->m_next_session_id++; - auto session = make(); - session->encoded_data = move(result.encoded_data); - session->decoder = move(result.decoder); - session->frame_count = result.frame_count; - strong_this->m_animation_sessions.set(session_id, move(session)); - } + main_thread_event_loop.deferred_invoke([strong_this = move(strong_this), job = move(job), request_id, result = move(result)] mutable { + auto current_job = strong_this->m_pending_jobs.get(request_id); + if (!current_job.has_value() || current_job.value() != job.ptr()) + return; - strong_this->async_did_decode_image(request_id, result.is_animated, result.loop_count, move(result.bitmaps), move(result.durations), result.scale, move(result.color_profile), session_id); - strong_this->m_pending_jobs.remove(request_id); - }, - [strong_this = NonnullRefPtr(*this), request_id](Error error) { - if (strong_this->is_open()) - strong_this->async_did_fail_to_decode_image(request_id, MUST(String::formatted("Decoding failed: {}", error))); - strong_this->m_pending_jobs.remove(request_id); + if (job->is_canceled()) { + strong_this->m_pending_jobs.remove(request_id); + return; + } + + if (result.is_error()) { + if (strong_this->is_open()) + strong_this->async_did_fail_to_decode_image(request_id, MUST(String::formatted("Decoding failed: {}", result.release_error()))); + strong_this->m_pending_jobs.remove(request_id); + return; + } + + auto result_value = result.release_value(); + i64 session_id = 0; + + if (result_value.decoder) { + // This is a streaming animated decode. Create a session. + session_id = strong_this->m_next_session_id++; + auto session = make_ref_counted(); + session->encoded_data = move(result_value.encoded_data); + session->decoder = move(result_value.decoder); + session->frame_count = result_value.frame_count; + strong_this->m_animation_sessions.set(session_id, move(session)); + } + + strong_this->async_did_decode_image(request_id, result_value.is_animated, result_value.loop_count, move(result_value.bitmaps), move(result_value.durations), result_value.scale, move(result_value.color_profile), session_id); + strong_this->m_pending_jobs.remove(request_id); + }); }); + + return job; } void ConnectionFromClient::decode_image(Core::AnonymousBuffer encoded_buffer, Optional ideal_size, Optional mime_type, i64 request_id) @@ -213,13 +233,13 @@ void ConnectionFromClient::decode_image(Core::AnonymousBuffer encoded_buffer, Op return; } - auto set_result = m_pending_jobs.set(request_id, make_decode_image_job(request_id, move(encoded_buffer), ideal_size, move(mime_type)), AK::HashSetExistingEntryBehavior::Keep); - - if (set_result != HashSetResult::InsertedNewEntry) { + if (m_pending_jobs.contains(request_id)) { m_pending_jobs.take(request_id).value()->cancel(); did_misbehave("Duplicate decode request id"); return; } + + m_pending_jobs.set(request_id, start_decode_image_job(request_id, move(encoded_buffer), ideal_size, move(mime_type))); } void ConnectionFromClient::cancel_decoding(i64 request_id) @@ -235,41 +255,75 @@ void ConnectionFromClient::request_animation_frames(i64 session_id, u32 start_fr if (it == m_animation_sessions.end()) return; - auto& session = *it->value; - auto decoder = session.decoder; - u32 const frame_count = session.frame_count; + auto session = it->value; + u32 const frame_count = session->frame_count; if (start_frame_index >= frame_count) return; u32 const end_index = min(frame_count, start_frame_index + min(count, frame_count - start_frame_index)); - auto job = FrameDecodeJob::construct( - [decoder, start_frame_index, end_index](auto&) -> ErrorOr> { - Vector frames; - frames.ensure_capacity(end_index - start_frame_index); - for (u32 i = start_frame_index; i < end_index; ++i) { - auto frame = TRY(decoder->frame(i)); - frame.image->set_alpha_type_destructive(Gfx::AlphaType::Premultiplied); - frames.unchecked_append(move(frame)); - } - return frames; - }, - [strong_this = NonnullRefPtr(*this), session_id](Vector frames) { - Vector> bitmaps; - bitmaps.ensure_capacity(frames.size()); - for (auto& frame : frames) - bitmaps.unchecked_append(move(frame.image)); - strong_this->async_did_decode_animation_frames(session_id, Gfx::BitmapSequence { move(bitmaps) }); - strong_this->m_pending_frame_jobs.remove(session_id); - }, - [strong_this = NonnullRefPtr(*this), session_id](Error error) { - if (strong_this->is_open()) - strong_this->async_did_fail_animation_decode(session_id, MUST(String::formatted("Frame decode failed: {}", error))); - strong_this->m_pending_frame_jobs.remove(session_id); + if (auto previous_job = m_pending_frame_jobs.take(session_id); previous_job.has_value()) + previous_job.value()->cancel(); + + m_pending_frame_jobs.set(session_id, start_frame_decode_job(session_id, move(session), start_frame_index, end_index)); +} + +NonnullRefPtr ConnectionFromClient::start_frame_decode_job(i64 session_id, NonnullRefPtr session, u32 start_frame_index, u32 end_index) +{ + auto job = make_ref_counted(); + auto& main_thread_event_loop = Core::EventLoop::current(); + Threading::ThreadPool::the().submit( + [strong_this = NonnullRefPtr(*this), job, session = move(session), &main_thread_event_loop, session_id, start_frame_index, end_index]() mutable { + auto result = [&]() -> ErrorOr { + if (job->is_canceled()) + return FrameDecodeResult {}; + + Sync::MutexLocker locker { session->decoder_mutex }; + if (!session->decoder) + return Error::from_string_literal("Animation session has no decoder"); + + Vector frames; + frames.ensure_capacity(end_index - start_frame_index); + for (u32 i = start_frame_index; i < end_index; ++i) { + if (job->is_canceled()) + return FrameDecodeResult {}; + + auto frame = TRY(session->decoder->frame(i)); + frame.image->set_alpha_type_destructive(Gfx::AlphaType::Premultiplied); + frames.unchecked_append(move(frame)); + } + return frames; + }(); + + main_thread_event_loop.deferred_invoke([strong_this = move(strong_this), job = move(job), session_id, result = move(result)] mutable { + auto current_job = strong_this->m_pending_frame_jobs.get(session_id); + if (!current_job.has_value() || current_job.value() != job.ptr()) + return; + + if (job->is_canceled()) { + strong_this->m_pending_frame_jobs.remove(session_id); + return; + } + + if (result.is_error()) { + if (strong_this->is_open()) + strong_this->async_did_fail_animation_decode(session_id, MUST(String::formatted("Frame decode failed: {}", result.release_error()))); + strong_this->m_pending_frame_jobs.remove(session_id); + return; + } + + auto frames = result.release_value(); + Vector> bitmaps; + bitmaps.ensure_capacity(frames.size()); + for (auto& frame : frames) + bitmaps.unchecked_append(move(frame.image)); + strong_this->async_did_decode_animation_frames(session_id, Gfx::BitmapSequence { move(bitmaps) }); + strong_this->m_pending_frame_jobs.remove(session_id); + }); }); - m_pending_frame_jobs.set(session_id, move(job)); + return job; } void ConnectionFromClient::stop_animation_decode(i64 session_id) diff --git a/Services/ImageDecoder/ConnectionFromClient.h b/Services/ImageDecoder/ConnectionFromClient.h index 88b33b5957..690fc0b6ef 100644 --- a/Services/ImageDecoder/ConnectionFromClient.h +++ b/Services/ImageDecoder/ConnectionFromClient.h @@ -6,6 +6,8 @@ #pragma once +#include +#include #include #include #include @@ -15,7 +17,7 @@ #include #include #include -#include +#include namespace ImageDecoder { @@ -42,16 +44,23 @@ public: Core::AnonymousBuffer encoded_data; }; - struct AnimationSession { + struct AnimationSession : public AtomicRefCounted { Core::AnonymousBuffer encoded_data; RefPtr decoder; u32 frame_count { 0 }; + Sync::Mutex decoder_mutex; }; private: - using Job = Threading::BackgroundAction; + struct PendingJob : public AtomicRefCounted { + void cancel() { m_canceled.store(true, AK::MemoryOrder::memory_order_relaxed); } + bool is_canceled() const { return m_canceled.load(AK::MemoryOrder::memory_order_relaxed); } + + private: + Atomic m_canceled { false }; + }; + using FrameDecodeResult = Vector; - using FrameDecodeJob = Threading::BackgroundAction; explicit ConnectionFromClient(NonnullOwnPtr); @@ -64,12 +73,13 @@ private: ErrorOr connect_new_client(); - NonnullRefPtr make_decode_image_job(i64 request_id, Core::AnonymousBuffer, Optional ideal_size, Optional mime_type); + NonnullRefPtr start_decode_image_job(i64 request_id, Core::AnonymousBuffer, Optional ideal_size, Optional mime_type); + NonnullRefPtr start_frame_decode_job(i64 session_id, NonnullRefPtr, u32 start_frame_index, u32 end_index); i64 m_next_session_id { 1 }; - HashMap> m_pending_jobs; - HashMap> m_animation_sessions; - HashMap> m_pending_frame_jobs; + HashMap> m_pending_jobs; + HashMap> m_animation_sessions; + HashMap> m_pending_frame_jobs; }; }