TaskWorklet: Tweak passing results from completed prerequisites

1. When selecting a thread for a task, consider which thread all
   prerequisites were assigned, not just the ones still pending.
2. Store the v8::Value for completed tasks.
3. If a task will run on the same thread as a completed
   prerequisite, hop to the worker thread and pass the
   prerequisite's result directly without a deserialization.
   (2) enables this, and (1) increases the frequency in which
   this optimization can be used.
4. Given that it is no longer guaranteed that the serialized
   result will be needed (i.e., if all dependents run on
   the same thread and task.result is never requested on the
   main thread, only the v8::Value will be used), move to a
   lazy serialization model, where the result is only serialized
   when it is promised via task.result or when a dependent is
   assigned to a different thread.

Bug: 879306
Change-Id: I22fe5f201f22b376861a5fb6d6e881ce0d5785cc
Reviewed-on: https://chromium-review.googlesource.com/c/1311498
Commit-Queue: Nate Chapin <japhet@chromium.org>
Reviewed-by: Hiroki Nakagawa <nhiroki@chromium.org>
Cr-Commit-Position: refs/heads/master@{#610136}

third_party/blink/renderer/core/workers/experimental/task.h

@@ -12,82 +12,93 @@
 #include "third_party/blink/renderer/platform/wtf/threading_primitives.h"
 
 namespace blink {
+class ResolveTask;
 class SerializedScriptValue;
 
 // Runs |function| with |arguments| on a thread from the given ThreadPool.
 // Scans |arguments| for Task objects, and registers those as dependencies,
 // passing the result of those tasks in place of the Task arguments.
 // All public functions are main-thread-only.
-// Task keeps itself alive via a SelfKeepAlive until the
+// TaskBase keeps itself alive via a SelfKeepAlive until the
 // the task completes and reports itself done on the main thread via
 // TaskCompleted().
-class Task final : public ScriptWrappable {
-  DEFINE_WRAPPERTYPEINFO();
-
+class TaskBase : public GarbageCollectedMixin {
  public:
-  // Called on main thread
-  Task(ThreadPoolThreadProvider*,
-       ScriptState*,
-       const ScriptValue& function,
-       const Vector<ScriptValue>& arguments,
-       TaskType);
-  Task(ThreadPoolThreadProvider*,
-       ScriptState*,
-       const String& function_name,
-       const Vector<ScriptValue>& arguments,
-       TaskType);
-  ~Task() override;
-  // Returns a promise that will be resolved with the result when it completes.
-  ScriptPromise result();
-  void cancel() LOCKS_EXCLUDED(mutex_);
+  virtual ~TaskBase();
 
-  void Trace(Visitor*) override;
+ protected:
+  virtual void StartTaskOnWorkerThread() LOCKS_EXCLUDED(mutex_) = 0;
+  virtual bool IsTargetThreadForArguments() = 0;
 
- private:
   enum class State { kPending, kStarted, kCancelPending, kCompleted, kFailed };
 
-  Task(ThreadPoolThreadProvider*,
-       ScriptState*,
-       const ScriptValue& function,
-       const String& function_name,
-       const Vector<ScriptValue>& arguments,
-       TaskType);
+  TaskBase(TaskType,
+           ScriptState*,
+           const ScriptValue& function,
+           const String& function_name);
+
+  void InitializeArgumentsOnMainThread(ThreadPoolThreadProvider*,
+                                       ScriptState*,
+                                       const Vector<ScriptValue>& arguments);
 
   class AsyncFunctionCompleted;
 
-  void StartTaskOnWorkerThread() LOCKS_EXCLUDED(mutex_);
+  // This caches the result after the task completes on the worker thread.
+  // We can't safely clear the ScopedPersistent from the main thread, so
+  // this wrappper allows us to hold a CrossThreadPersistent that arranges
+  // for GC on the worker thread.
+  class V8ResultHolder final
+      : public GarbageCollectedFinalized<V8ResultHolder> {
+   public:
+    V8ResultHolder(v8::Isolate* isolate, v8::Local<v8::Value> result)
+        : result_(isolate, result) {}
+    ~V8ResultHolder() = default;
+    v8::Local<v8::Value> GetResult(v8::Isolate* isolate) {
+      return result_.NewLocal(isolate);
+    }
+    void Trace(Visitor*) {}
+
+   private:
+    ScopedPersistent<v8::Value> result_;
+  };
+
+  bool WillStartTaskOnWorkerThread();
   void RunTaskOnWorkerThread();
-  void TaskCompletedOnWorkerThread(v8::Local<v8::Value> return_value, State)
+  void TaskCompletedOnWorkerThread(v8::Local<v8::Value> v8_result, State)
+      LOCKS_EXCLUDED(mutex_);
+  void PassResultToDependentOnWorkerThread(size_t dependent_index, TaskBase*)
       LOCKS_EXCLUDED(mutex_);
 
   // Called on ANY thread (main thread, worker_thread_, or a sibling worker).
   void MaybeStartTask() EXCLUSIVE_LOCKS_REQUIRED(mutex_);
-  void PrerequisiteFinished(size_t prerequisite_index,
-                            v8::Local<v8::Value>,
+  void PrerequisiteFinished(size_t index,
+                            V8ResultHolder*,
                             scoped_refptr<SerializedScriptValue>,
-                            State) LOCKS_EXCLUDED(mutex_);
+                            bool failed) LOCKS_EXCLUDED(mutex_);
   bool HasFinished() const EXCLUSIVE_LOCKS_REQUIRED(mutex_) {
     return state_ == State::kCompleted || state_ == State::kFailed;
   }
   void AdvanceState(State new_state) EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
+  // Called on main thread or worker_thread_
+  scoped_refptr<SerializedScriptValue> GetSerializedResult()
+      LOCKS_EXCLUDED(mutex_);
+
   // Called on main thread
   static ThreadPoolThread* SelectThread(
-      const HeapVector<Member<Task>>& prerequisites,
+      const HeapVector<Member<TaskBase>>& prerequisites,
       ThreadPoolThreadProvider*);
-  ThreadPoolThread* GetScheduledThread() LOCKS_EXCLUDED(mutex_);
-  void RegisterDependencies(const HeapVector<Member<Task>>& prerequisites,
+  void RegisterDependencies(const HeapVector<Member<TaskBase>>& prerequisites,
                             const Vector<size_t>& prerequisite_indices)
       LOCKS_EXCLUDED(mutex_);
-  void TaskCompleted();
+  virtual void TaskCompleted(bool was_successful);
 
   // worker_thread_ is selected in the constructor and not changed thereafter.
   ThreadPoolThread* worker_thread_ = nullptr;
   const TaskType task_type_;
 
   // Main thread only
-  SelfKeepAlive<Task> self_keep_alive_;
-  Member<ScriptPromiseResolver> resolver_;
+  SelfKeepAlive<TaskBase> self_keep_alive_;
 
   // Created in constructor on the main thread, consumed and cleared on
   // worker_thread_. Those steps can't overlap, so no mutex_ required.
@@ -97,20 +108,23 @@ class Task final : public ScriptWrappable {
   // Created and populated with non-prerequiste parameters on the main thread.
   // Each prerequisite writes its return value into arguments_ from its thread.
   // If the prequisite and this have the same worker_thread_, there is no need
-  // to serialize and deserialize the argument, so v8_argument will be populated
+  // to serialize and deserialize the argument, so v8_value will be populated
   // with the v8::Value returned by the prerequisite.
   // Consumed and cleared on worker_thread_.
   // Only requires mutex_ when writing prerequisite results, at other times
   // either the main thread or the worker_thread_ has sole access.
   struct Argument {
     scoped_refptr<SerializedScriptValue> serialized_value;
-    std::unique_ptr<ScopedPersistent<v8::Value>> v8_value;
+    CrossThreadPersistent<V8ResultHolder> v8_value;
   };
   Vector<Argument> arguments_;
 
   // Read on main thread, write on worker_thread_.
   scoped_refptr<SerializedScriptValue> serialized_result_ GUARDED_BY(mutex_);
 
+  // Read/write on worker_thread_
+  CrossThreadPersistent<V8ResultHolder> v8_result_;
+
   // Read/write on both main thread and worker_thread_.
   State state_ GUARDED_BY(mutex_) = State::kPending;
 
@@ -122,11 +136,11 @@ class Task final : public ScriptWrappable {
   // Each element in dependents_ is not yet in the kCompleted state.
   struct Dependent final : public GarbageCollected<Dependent> {
    public:
-    Dependent(Task* task, size_t index) : task(task), index(index) {
+    Dependent(TaskBase* task, size_t index) : task(task), index(index) {
       DCHECK(IsMainThread());
     }
     void Trace(Visitor* visitor) { visitor->Trace(task); }
-    Member<Task> task;
+    Member<TaskBase> task;
     // The index in the dependent's argument array where this result should go.
     size_t index;
   };
@@ -135,6 +149,65 @@ class Task final : public ScriptWrappable {
   Mutex mutex_;
 };
 
+// The variant of TaskBase that is exposed to JS.
+class Task final : public ScriptWrappable, public TaskBase {
+  DEFINE_WRAPPERTYPEINFO();
+  USING_GARBAGE_COLLECTED_MIXIN(Task);
+
+ public:
+  // Called on main thread
+  Task(ThreadPoolThreadProvider* thread_provider,
+       ScriptState* script_state,
+       const ScriptValue& function,
+       const Vector<ScriptValue>& arguments,
+       TaskType task_type)
+      : TaskBase(task_type, script_state, function, String()) {
+    InitializeArgumentsOnMainThread(thread_provider, script_state, arguments);
+  }
+  Task(ThreadPoolThreadProvider* thread_provider,
+       ScriptState* script_state,
+       const String& function_name,
+       const Vector<ScriptValue>& arguments,
+       TaskType task_type)
+      : TaskBase(task_type, script_state, ScriptValue(), function_name) {
+    InitializeArgumentsOnMainThread(thread_provider, script_state, arguments);
+  }
+
+  // Returns a promise that will be resolved with the result when it completes.
+  ScriptPromise result(ScriptState*);
+  void cancel() LOCKS_EXCLUDED(mutex_);
+
+  void StartTaskOnWorkerThread() override LOCKS_EXCLUDED(mutex_);
+  bool IsTargetThreadForArguments() override {
+    return worker_thread_->IsCurrentThread();
+  }
+  void Trace(Visitor*) override;
+
+ private:
+  Member<ResolveTask> resolve_task_;
+};
+
+// An internal TaskBase subclass that drives main thread promise resolution.
+// It registers itself as a dependent on the Task whose result is being
+// promised. When that Task completes, it runs a dummy script that just returns
+// the dependent's result as its own. It then eagerly serializes the result, and
+// overrides TaskCompleted() to actually resolve the promise.
+class ResolveTask final : public GarbageCollectedFinalized<ResolveTask>,
+                          public TaskBase {
+  USING_GARBAGE_COLLECTED_MIXIN(ResolveTask);
+
+ public:
+  ResolveTask(ScriptState*, TaskType, Task* prerequisite);
+  void StartTaskOnWorkerThread() override LOCKS_EXCLUDED(mutex_);
+  bool IsTargetThreadForArguments() override { return IsMainThread(); }
+  void TaskCompleted(bool was_successful) override;
+  ScriptPromise GetPromise() { return resolver_->Promise(); }
+  void Trace(Visitor*) override;
+
+ private:
+  Member<ScriptPromiseResolver> resolver_;
+};
+
 }  // namespace blink
 
 #endif  // THIRD_PARTY_BLINK_RENDERER_CORE_WORKERS_EXPERIMENTAL_TASK_H_

third_party/blink/renderer/core/workers/experimental/task.idl

@@ -6,6 +6,6 @@
     Exposed=Window,
     RuntimeEnabled=WorkerTaskQueue
 ] interface Task {
-    readonly attribute Promise<any> result;
+    [CallWith=ScriptState] readonly attribute Promise<any> result;
     void cancel();
 };

third_party/blink/renderer/core/workers/experimental/worker_task_queue.cc

@@ -49,7 +49,7 @@ ScriptPromise WorkerTaskQueue::postFunction(
                         arguments, task_type_);
   if (signal)
     signal->AddAlgorithm(WTF::Bind(&Task::cancel, WrapWeakPersistent(task)));
-  return task->result();
+  return task->result(script_state);
 }
 
 Task* WorkerTaskQueue::postTask(ScriptState* script_state,