mirror of
https://github.com/LadybirdBrowser/ladybird.git
synced 2025-07-24 18:02:20 +00:00
LibWeb: Add "parallel queue" and allow it as fetch task destination
Note that it's not actually executing tasks in parallel, it's still throwing them on the HTML event loop task queue, each with its own unique task source. This makes our fetch implementation a lot more robust when HTTP caching is enabled, and you can now click links on https://terminal.shop/ without hitting TODO assertions in fetch.
This commit is contained in:
parent
9a5ef95022
commit
03256a2543
Notes:
github-actions[bot]
2025-07-16 22:14:47 +00:00
Author: https://github.com/awesomekling
Commit: 03256a2543
Pull-request: https://github.com/LadybirdBrowser/ladybird/pull/5471
10 changed files with 65 additions and 44 deletions
|
@ -68,22 +68,21 @@ void Body::fully_read(JS::Realm& realm, Web::Fetch::Infrastructure::Body::Proces
|
|||
{
|
||||
HTML::TemporaryExecutionContext execution_context { realm, HTML::TemporaryExecutionContext::CallbacksEnabled::Yes };
|
||||
|
||||
// FIXME: 1. If taskDestination is null, then set taskDestination to the result of starting a new parallel queue.
|
||||
// FIXME: Handle 'parallel queue' task destination
|
||||
VERIFY(!task_destination.has<Empty>());
|
||||
auto task_destination_object = task_destination.get<GC::Ref<JS::Object>>();
|
||||
// 1. If taskDestination is null, then set taskDestination to the result of starting a new parallel queue.
|
||||
if (task_destination.has<Empty>())
|
||||
task_destination = HTML::ParallelQueue::create();
|
||||
|
||||
// 2. Let successSteps given a byte sequence bytes be to queue a fetch task to run processBody given bytes, with taskDestination.
|
||||
auto success_steps = [&realm, process_body, task_destination_object](ByteBuffer bytes) {
|
||||
queue_fetch_task(*task_destination_object, GC::create_function(realm.heap(), [process_body, bytes = move(bytes)]() mutable {
|
||||
auto success_steps = [&realm, process_body, task_destination](ByteBuffer bytes) {
|
||||
queue_fetch_task(task_destination, GC::create_function(realm.heap(), [process_body, bytes = move(bytes)]() mutable {
|
||||
process_body->function()(move(bytes));
|
||||
}));
|
||||
};
|
||||
|
||||
// 3. Let errorSteps optionally given an exception exception be to queue a fetch task to run processBodyError given
|
||||
// exception, with taskDestination.
|
||||
auto error_steps = [&realm, process_body_error, task_destination_object](JS::Value exception) {
|
||||
queue_fetch_task(*task_destination_object, GC::create_function(realm.heap(), [process_body_error, exception]() {
|
||||
auto error_steps = [&realm, process_body_error, task_destination](JS::Value exception) {
|
||||
queue_fetch_task(task_destination, GC::create_function(realm.heap(), [process_body_error, exception]() {
|
||||
process_body_error->function()(exception);
|
||||
}));
|
||||
};
|
||||
|
@ -107,20 +106,21 @@ void Body::incrementally_read(ProcessBodyChunkCallback process_body_chunk, Proce
|
|||
{
|
||||
HTML::TemporaryExecutionContext const execution_context { m_stream->realm(), HTML::TemporaryExecutionContext::CallbacksEnabled::Yes };
|
||||
|
||||
VERIFY(task_destination.has<GC::Ref<JS::Object>>());
|
||||
// FIXME: 1. If taskDestination is null, then set taskDestination to the result of starting a new parallel queue.
|
||||
// FIXME: Handle 'parallel queue' task destination
|
||||
// 1. If taskDestination is null, then set taskDestination to the result of starting a new parallel queue.
|
||||
if (task_destination.has<Empty>())
|
||||
task_destination = HTML::ParallelQueue::create();
|
||||
|
||||
// 2. Let reader be the result of getting a reader for body’s stream.
|
||||
// NOTE: This operation will not throw an exception.
|
||||
auto reader = MUST(m_stream->get_a_reader());
|
||||
|
||||
// 3. Perform the incrementally-read loop given reader, taskDestination, processBodyChunk, processEndOfBody, and processBodyError.
|
||||
VERIFY(!task_destination.has<Empty>());
|
||||
incrementally_read_loop(reader, task_destination.get<GC::Ref<JS::Object>>(), process_body_chunk, process_end_of_body, process_body_error);
|
||||
}
|
||||
|
||||
// https://fetch.spec.whatwg.org/#incrementally-read-loop
|
||||
void Body::incrementally_read_loop(Streams::ReadableStreamDefaultReader& reader, GC::Ref<JS::Object> task_destination, ProcessBodyChunkCallback process_body_chunk, ProcessEndOfBodyCallback process_end_of_body, ProcessBodyErrorCallback process_body_error)
|
||||
void Body::incrementally_read_loop(Streams::ReadableStreamDefaultReader& reader, TaskDestination task_destination, ProcessBodyChunkCallback process_body_chunk, ProcessEndOfBodyCallback process_end_of_body, ProcessBodyErrorCallback process_body_error)
|
||||
{
|
||||
auto& realm = reader.realm();
|
||||
// 1. Let readRequest be the following read request:
|
||||
|
|
|
@ -46,9 +46,9 @@ public:
|
|||
|
||||
[[nodiscard]] GC::Ref<Body> clone(JS::Realm&);
|
||||
|
||||
void fully_read(JS::Realm&, ProcessBodyCallback process_body, ProcessBodyErrorCallback process_body_error, TaskDestination task_destination) const;
|
||||
void incrementally_read(ProcessBodyChunkCallback process_body_chunk, ProcessEndOfBodyCallback process_end_of_body, ProcessBodyErrorCallback process_body_error, TaskDestination task_destination);
|
||||
void incrementally_read_loop(Streams::ReadableStreamDefaultReader& reader, GC::Ref<JS::Object> task_destination, ProcessBodyChunkCallback process_body_chunk, ProcessEndOfBodyCallback process_end_of_body, ProcessBodyErrorCallback process_body_error);
|
||||
void fully_read(JS::Realm&, ProcessBodyCallback process_body, ProcessBodyErrorCallback process_body_error, TaskDestination) const;
|
||||
void incrementally_read(ProcessBodyChunkCallback process_body_chunk, ProcessEndOfBodyCallback process_end_of_body, ProcessBodyErrorCallback process_body_error, TaskDestination);
|
||||
void incrementally_read_loop(Streams::ReadableStreamDefaultReader& reader, TaskDestination, ProcessBodyChunkCallback process_body_chunk, ProcessEndOfBodyCallback process_end_of_body, ProcessBodyErrorCallback process_body_error);
|
||||
|
||||
virtual void visit_edges(JS::Cell::Visitor&) override;
|
||||
|
||||
|
|
|
@ -61,7 +61,7 @@ void IncrementalReadLoopReadRequest::on_error(JS::Value error)
|
|||
}));
|
||||
}
|
||||
|
||||
IncrementalReadLoopReadRequest::IncrementalReadLoopReadRequest(GC::Ref<Body> body, GC::Ref<Streams::ReadableStreamDefaultReader> reader, GC::Ref<JS::Object> task_destination, Body::ProcessBodyChunkCallback process_body_chunk, Body::ProcessEndOfBodyCallback process_end_of_body, Body::ProcessBodyErrorCallback process_body_error)
|
||||
IncrementalReadLoopReadRequest::IncrementalReadLoopReadRequest(GC::Ref<Body> body, GC::Ref<Streams::ReadableStreamDefaultReader> reader, TaskDestination task_destination, Body::ProcessBodyChunkCallback process_body_chunk, Body::ProcessEndOfBodyCallback process_end_of_body, Body::ProcessBodyErrorCallback process_body_error)
|
||||
: m_body(body)
|
||||
, m_reader(reader)
|
||||
, m_task_destination(task_destination)
|
||||
|
@ -76,7 +76,8 @@ void IncrementalReadLoopReadRequest::visit_edges(Visitor& visitor)
|
|||
Base::visit_edges(visitor);
|
||||
visitor.visit(m_body);
|
||||
visitor.visit(m_reader);
|
||||
visitor.visit(m_task_destination);
|
||||
if (auto* task_destination_object = m_task_destination.get_pointer<GC::Ref<JS::Object>>(); task_destination_object)
|
||||
visitor.visit(*task_destination_object);
|
||||
visitor.visit(m_process_body_chunk);
|
||||
visitor.visit(m_process_end_of_body);
|
||||
visitor.visit(m_process_body_error);
|
||||
|
|
|
@ -17,7 +17,7 @@ class IncrementalReadLoopReadRequest : public Streams::ReadRequest {
|
|||
GC_DECLARE_ALLOCATOR(IncrementalReadLoopReadRequest);
|
||||
|
||||
public:
|
||||
IncrementalReadLoopReadRequest(GC::Ref<Body>, GC::Ref<Streams::ReadableStreamDefaultReader>, GC::Ref<JS::Object> task_destination, Body::ProcessBodyChunkCallback, Body::ProcessEndOfBodyCallback, Body::ProcessBodyErrorCallback);
|
||||
IncrementalReadLoopReadRequest(GC::Ref<Body>, GC::Ref<Streams::ReadableStreamDefaultReader>, TaskDestination, Body::ProcessBodyChunkCallback, Body::ProcessEndOfBodyCallback, Body::ProcessBodyErrorCallback);
|
||||
|
||||
virtual void on_chunk(JS::Value chunk) override;
|
||||
virtual void on_close() override;
|
||||
|
@ -28,7 +28,7 @@ private:
|
|||
|
||||
GC::Ref<Body> m_body;
|
||||
GC::Ref<Streams::ReadableStreamDefaultReader> m_reader;
|
||||
GC::Ref<JS::Object> m_task_destination;
|
||||
TaskDestination m_task_destination;
|
||||
Body::ProcessBodyChunkCallback m_process_body_chunk;
|
||||
Body::ProcessEndOfBodyCallback m_process_end_of_body;
|
||||
Body::ProcessBodyErrorCallback m_process_body_error;
|
||||
|
|
|
@ -4,6 +4,7 @@
|
|||
* SPDX-License-Identifier: BSD-2-Clause
|
||||
*/
|
||||
|
||||
#include <LibWeb/Bindings/MainThreadVM.h>
|
||||
#include <LibWeb/Fetch/Infrastructure/FetchController.h>
|
||||
#include <LibWeb/Fetch/Infrastructure/Task.h>
|
||||
#include <LibWeb/HTML/EventLoop/EventLoop.h>
|
||||
|
@ -11,21 +12,25 @@
|
|||
namespace Web::Fetch::Infrastructure {
|
||||
|
||||
// https://fetch.spec.whatwg.org/#queue-a-fetch-task
|
||||
HTML::TaskID queue_fetch_task(JS::Object& task_destination, GC::Ref<GC::Function<void()>> algorithm)
|
||||
HTML::TaskID queue_fetch_task(TaskDestination task_destination, GC::Ref<GC::Function<void()>> algorithm)
|
||||
{
|
||||
// FIXME: 1. If taskDestination is a parallel queue, then enqueue algorithm to taskDestination.
|
||||
VERIFY(!task_destination.has<Empty>());
|
||||
|
||||
// 1. If taskDestination is a parallel queue, then enqueue algorithm to taskDestination.
|
||||
if (auto* parallel_queue = task_destination.get_pointer<NonnullRefPtr<HTML::ParallelQueue>>())
|
||||
return (*parallel_queue)->enqueue(algorithm);
|
||||
|
||||
// 2. Otherwise, queue a global task on the networking task source with taskDestination and algorithm.
|
||||
return HTML::queue_global_task(HTML::Task::Source::Networking, task_destination, algorithm);
|
||||
return HTML::queue_global_task(HTML::Task::Source::Networking, task_destination.get<GC::Ref<JS::Object>>(), algorithm);
|
||||
}
|
||||
|
||||
// AD-HOC: This overload allows tracking the queued task within the fetch controller so that we may cancel queued tasks
|
||||
// when the spec indicates that we must stop an ongoing fetch.
|
||||
HTML::TaskID queue_fetch_task(GC::Ref<FetchController> fetch_controller, JS::Object& task_destination, GC::Ref<GC::Function<void()>> algorithm)
|
||||
HTML::TaskID queue_fetch_task(GC::Ref<FetchController> fetch_controller, TaskDestination task_destination, GC::Ref<GC::Function<void()>> algorithm)
|
||||
{
|
||||
auto fetch_task_id = fetch_controller->next_fetch_task_id();
|
||||
|
||||
auto& heap = task_destination.heap();
|
||||
auto& heap = fetch_controller->heap();
|
||||
auto html_task_id = queue_fetch_task(task_destination, GC::create_function(heap, [fetch_controller, fetch_task_id, algorithm]() {
|
||||
fetch_controller->fetch_task_complete(fetch_task_id);
|
||||
algorithm->function()();
|
||||
|
|
|
@ -14,10 +14,9 @@
|
|||
|
||||
namespace Web::Fetch::Infrastructure {
|
||||
|
||||
// FIXME: 'or a parallel queue'
|
||||
using TaskDestination = Variant<Empty, GC::Ref<JS::Object>>;
|
||||
using TaskDestination = Variant<Empty, GC::Ref<JS::Object>, NonnullRefPtr<HTML::ParallelQueue>>;
|
||||
|
||||
HTML::TaskID queue_fetch_task(JS::Object&, GC::Ref<GC::Function<void()>>);
|
||||
HTML::TaskID queue_fetch_task(GC::Ref<FetchController>, JS::Object&, GC::Ref<GC::Function<void()>>);
|
||||
HTML::TaskID queue_fetch_task(TaskDestination, GC::Ref<GC::Function<void()>>);
|
||||
HTML::TaskID queue_fetch_task(GC::Ref<FetchController>, TaskDestination, GC::Ref<GC::Function<void()>>);
|
||||
|
||||
}
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue