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.
This commit is contained in:
Andreas Kling
Andreas Kling
parent
540827040c
commit
88dfee5528
2 changed files with 126 additions and 62 deletions
|
|
@ -6,14 +6,18 @@
|
|||
|
||||
#include <AK/Debug.h>
|
||||
#include <AK/IDAllocator.h>
|
||||
#include <AK/NonnullRefPtr.h>
|
||||
#include <ImageDecoder/ConnectionFromClient.h>
|
||||
#include <ImageDecoder/ImageDecoderClientEndpoint.h>
|
||||
#include <LibCore/EventLoop.h>
|
||||
#include <LibCore/Process.h>
|
||||
#include <LibCore/System.h>
|
||||
#include <LibGfx/Bitmap.h>
|
||||
#include <LibGfx/ImageFormats/ImageDecoder.h>
|
||||
#include <LibGfx/ImageFormats/TIFFMetadata.h>
|
||||
#include <LibIPC/TransportHandle.h>
|
||||
#include <LibSync/Mutex.h>
|
||||
#include <LibThreading/ThreadPool.h>
|
||||
|
||||
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<ConnectionFromClient::DecodeResult> decode_image_to_details(Core:
|
|||
return result;
|
||||
}
|
||||
|
||||
NonnullRefPtr<ConnectionFromClient::Job> ConnectionFromClient::make_decode_image_job(i64 request_id, Core::AnonymousBuffer encoded_buffer, Optional<Gfx::IntSize> ideal_size, Optional<ByteString> mime_type)
|
||||
NonnullRefPtr<ConnectionFromClient::PendingJob> ConnectionFromClient::start_decode_image_job(i64 request_id, Core::AnonymousBuffer encoded_buffer, Optional<Gfx::IntSize> ideal_size, Optional<ByteString> mime_type)
|
||||
{
|
||||
return Job::construct(
|
||||
[encoded_buffer = move(encoded_buffer), ideal_size = move(ideal_size), mime_type = move(mime_type)](auto&) mutable -> ErrorOr<DecodeResult> {
|
||||
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<PendingJob>();
|
||||
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<AnimationSession>();
|
||||
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<AnimationSession>();
|
||||
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<Gfx::IntSize> ideal_size, Optional<ByteString> 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<Gfx::ImageFrameDescriptor>> {
|
||||
Vector<Gfx::ImageFrameDescriptor> 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<Gfx::ImageFrameDescriptor> frames) {
|
||||
Vector<RefPtr<Gfx::Bitmap>> 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::PendingJob> ConnectionFromClient::start_frame_decode_job(i64 session_id, NonnullRefPtr<AnimationSession> session, u32 start_frame_index, u32 end_index)
|
||||
{
|
||||
auto job = make_ref_counted<PendingJob>();
|
||||
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<FrameDecodeResult> {
|
||||
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<Gfx::ImageFrameDescriptor> 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<RefPtr<Gfx::Bitmap>> 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)
|
||||
|
|
|
|||
|
|
@ -6,6 +6,8 @@
|
|||
|
||||
#pragma once
|
||||
|
||||
#include <AK/Atomic.h>
|
||||
#include <AK/AtomicRefCounted.h>
|
||||
#include <AK/HashMap.h>
|
||||
#include <ImageDecoder/Forward.h>
|
||||
#include <ImageDecoder/ImageDecoderClientEndpoint.h>
|
||||
|
|
@ -15,7 +17,7 @@
|
|||
#include <LibGfx/ColorSpace.h>
|
||||
#include <LibGfx/ImageFormats/ImageDecoder.h>
|
||||
#include <LibIPC/ConnectionFromClient.h>
|
||||
#include <LibThreading/BackgroundAction.h>
|
||||
#include <LibSync/Mutex.h>
|
||||
|
||||
namespace ImageDecoder {
|
||||
|
||||
|
|
@ -42,16 +44,23 @@ public:
|
|||
Core::AnonymousBuffer encoded_data;
|
||||
};
|
||||
|
||||
struct AnimationSession {
|
||||
struct AnimationSession : public AtomicRefCounted<AnimationSession> {
|
||||
Core::AnonymousBuffer encoded_data;
|
||||
RefPtr<Gfx::ImageDecoder> decoder;
|
||||
u32 frame_count { 0 };
|
||||
Sync::Mutex decoder_mutex;
|
||||
};
|
||||
|
||||
private:
|
||||
using Job = Threading::BackgroundAction<DecodeResult>;
|
||||
struct PendingJob : public AtomicRefCounted<PendingJob> {
|
||||
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<bool> m_canceled { false };
|
||||
};
|
||||
|
||||
using FrameDecodeResult = Vector<Gfx::ImageFrameDescriptor>;
|
||||
using FrameDecodeJob = Threading::BackgroundAction<FrameDecodeResult>;
|
||||
|
||||
explicit ConnectionFromClient(NonnullOwnPtr<IPC::Transport>);
|
||||
|
||||
|
|
@ -64,12 +73,13 @@ private:
|
|||
|
||||
ErrorOr<IPC::TransportHandle> connect_new_client();
|
||||
|
||||
NonnullRefPtr<Job> make_decode_image_job(i64 request_id, Core::AnonymousBuffer, Optional<Gfx::IntSize> ideal_size, Optional<ByteString> mime_type);
|
||||
NonnullRefPtr<PendingJob> start_decode_image_job(i64 request_id, Core::AnonymousBuffer, Optional<Gfx::IntSize> ideal_size, Optional<ByteString> mime_type);
|
||||
NonnullRefPtr<PendingJob> start_frame_decode_job(i64 session_id, NonnullRefPtr<AnimationSession>, u32 start_frame_index, u32 end_index);
|
||||
|
||||
i64 m_next_session_id { 1 };
|
||||
HashMap<i64, NonnullRefPtr<Job>> m_pending_jobs;
|
||||
HashMap<i64, NonnullOwnPtr<AnimationSession>> m_animation_sessions;
|
||||
HashMap<i64, NonnullRefPtr<FrameDecodeJob>> m_pending_frame_jobs;
|
||||
HashMap<i64, NonnullRefPtr<PendingJob>> m_pending_jobs;
|
||||
HashMap<i64, NonnullRefPtr<AnimationSession>> m_animation_sessions;
|
||||
HashMap<i64, NonnullRefPtr<PendingJob>> m_pending_frame_jobs;
|
||||
};
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue