[d8] Add --streaming-compile flag for streamer tests

Adds a --streaming-compile flag to d8 so that we can test the streaming/
background compilation (e.g. for performance testing).

This differs from the --stress-background-compile flag in two main ways:

    1) It's not a stress test, so it doesn't run a main-thread compile
       for verification, and
    2) It uses the "proper" API, and (like Chromium) pumps the message
       loop while waiting for compilation to complete, so e.g. GC idle
       tasks can run.

Change-Id: I1ea1badf39d25076d95c8d19f173510da277541f
Reviewed-on: https://chromium-review.googlesource.com/c/v8/v8/+/2219937
Commit-Queue: Ulan Degenbaev <ulan@chromium.org>
Reviewed-by: Ulan Degenbaev <ulan@chromium.org>
Auto-Submit: Leszek Swirski <leszeks@chromium.org>
Cr-Commit-Position: refs/heads/master@{#68069}

src/d8/d8.cc

@@ -444,6 +444,7 @@ std::unordered_set<std::shared_ptr<Worker>> Shell::running_workers_;
 std::atomic<bool> Shell::script_executed_{false};
 base::LazyMutex Shell::isolate_status_lock_;
 std::map<v8::Isolate*, bool> Shell::isolate_status_;
+std::map<v8::Isolate*, int> Shell::isolate_running_streaming_tasks_;
 base::LazyMutex Shell::cached_code_mutex_;
 std::map<std::string, std::unique_ptr<ScriptCompiler::CachedData>>
     Shell::cached_code_map_;
@@ -486,6 +487,61 @@ void Shell::StoreInCodeCache(Isolate* isolate, Local<Value> source,
                                      ScriptCompiler::CachedData::BufferOwned));
 }
 
+// Dummy external source stream which returns the whole source in one go.
+// TODO(leszeks): Also test chunking the data.
+class DummySourceStream : public v8::ScriptCompiler::ExternalSourceStream {
+ public:
+  explicit DummySourceStream(Local<String> source) : done_(false) {
+    source_buffer_ = Utils::OpenHandle(*source)->ToCString(
+        i::ALLOW_NULLS, i::FAST_STRING_TRAVERSAL, &source_length_);
+  }
+
+  size_t GetMoreData(const uint8_t** src) override {
+    if (done_) {
+      return 0;
+    }
+    *src = reinterpret_cast<uint8_t*>(source_buffer_.release());
+    done_ = true;
+
+    return source_length_;
+  }
+
+ private:
+  int source_length_;
+  std::unique_ptr<char[]> source_buffer_;
+  bool done_;
+};
+
+class StreamingCompileTask final : public v8::Task {
+ public:
+  StreamingCompileTask(Isolate* isolate,
+                       v8::ScriptCompiler::StreamedSource* streamed_source)
+      : isolate_(isolate),
+        script_streaming_task_(v8::ScriptCompiler::StartStreamingScript(
+            isolate, streamed_source)) {
+    Shell::NotifyStartStreamingTask(isolate_);
+  }
+
+  void Run() override {
+    script_streaming_task_->Run();
+    // Signal that the task has finished using the task runner to wake the
+    // message loop.
+    Shell::PostForegroundTask(isolate_, std::make_unique<FinishTask>(isolate_));
+  }
+
+ private:
+  class FinishTask final : public v8::Task {
+   public:
+    explicit FinishTask(Isolate* isolate) : isolate_(isolate) {}
+    void Run() final { Shell::NotifyFinishStreamingTask(isolate_); }
+    Isolate* isolate_;
+  };
+
+  Isolate* isolate_;
+  std::unique_ptr<v8::ScriptCompiler::ScriptStreamingTask>
+      script_streaming_task_;
+};
+
 // Executes a string within the current v8 context.
 bool Shell::ExecuteString(Isolate* isolate, Local<String> source,
                           Local<Value> name, PrintResult print_result,
@@ -547,6 +603,19 @@ bool Shell::ExecuteString(Isolate* isolate, Local<String> source,
         maybe_script = ScriptCompiler::Compile(
             context, &script_source, ScriptCompiler::kNoCompileOptions);
       }
+    } else if (options.streaming_compile) {
+      v8::ScriptCompiler::StreamedSource streamed_source(
+          std::make_unique<DummySourceStream>(source),
+          v8::ScriptCompiler::StreamedSource::UTF8);
+
+      PostBlockingBackgroundTask(
+          std::make_unique<StreamingCompileTask>(isolate, &streamed_source));
+
+      // Pump the loop until the streaming task completes.
+      Shell::CompleteMessageLoop(isolate);
+
+      maybe_script =
+          ScriptCompiler::Compile(context, &streamed_source, source, origin);
     } else {
       ScriptCompiler::Source script_source(source, origin);
       maybe_script = ScriptCompiler::Compile(context, &script_source,
@@ -2936,6 +3005,13 @@ bool Shell::SetOptions(int argc, char* argv[]) {
         return false;
       }
       argv[i] = nullptr;
+    } else if (strcmp(argv[i], "--streaming-compile") == 0) {
+      options.streaming_compile = true;
+      argv[i] = nullptr;
+    } else if ((strcmp(argv[i], "--no-streaming-compile") == 0) ||
+               (strcmp(argv[i], "--nostreaming-compile") == 0)) {
+      options.streaming_compile = false;
+      argv[i] = nullptr;
     } else if (strcmp(argv[i], "--enable-tracing") == 0) {
       options.trace_enabled = true;
       argv[i] = nullptr;
@@ -3094,11 +3170,20 @@ void Shell::CollectGarbage(Isolate* isolate) {
 
 void Shell::SetWaitUntilDone(Isolate* isolate, bool value) {
   base::MutexGuard guard(isolate_status_lock_.Pointer());
-  if (isolate_status_.count(isolate) == 0) {
-    isolate_status_.insert(std::make_pair(isolate, value));
-  } else {
-    isolate_status_[isolate] = value;
-  }
+  isolate_status_[isolate] = value;
+}
+
+void Shell::NotifyStartStreamingTask(Isolate* isolate) {
+  DCHECK(options.streaming_compile);
+  base::MutexGuard guard(isolate_status_lock_.Pointer());
+  ++isolate_running_streaming_tasks_[isolate];
+}
+
+void Shell::NotifyFinishStreamingTask(Isolate* isolate) {
+  DCHECK(options.streaming_compile);
+  base::MutexGuard guard(isolate_status_lock_.Pointer());
+  --isolate_running_streaming_tasks_[isolate];
+  DCHECK_GE(isolate_running_streaming_tasks_[isolate], 0);
 }
 
 namespace {
@@ -3164,7 +3249,8 @@ bool Shell::CompleteMessageLoop(Isolate* isolate) {
     i::wasm::WasmEngine* wasm_engine = i_isolate->wasm_engine();
     bool should_wait = (options.wait_for_wasm &&
                         wasm_engine->HasRunningCompileJob(i_isolate)) ||
-                       isolate_status_[isolate];
+                       isolate_status_[isolate] ||
+                       isolate_running_streaming_tasks_[isolate] > 0;
     return should_wait ? platform::MessageLoopBehavior::kWaitForWork
                        : platform::MessageLoopBehavior::kDoNotWait;
   };
@@ -3176,6 +3262,15 @@ bool Shell::EmptyMessageQueues(Isolate* isolate) {
       isolate, []() { return platform::MessageLoopBehavior::kDoNotWait; });
 }
 
+void Shell::PostForegroundTask(Isolate* isolate, std::unique_ptr<Task> task) {
+  g_default_platform->GetForegroundTaskRunner(isolate)->PostTask(
+      std::move(task));
+}
+
+void Shell::PostBlockingBackgroundTask(std::unique_ptr<Task> task) {
+  g_default_platform->CallBlockingTaskOnWorkerThread(std::move(task));
+}
+
 class Serializer : public ValueSerializer::Delegate {
  public:
   explicit Serializer(Isolate* isolate)

src/d8/d8.h

@@ -280,6 +280,7 @@ class ShellOptions {
   v8::ScriptCompiler::CompileOptions compile_options =
       v8::ScriptCompiler::kNoCompileOptions;
   CodeCacheOptions code_cache_options = CodeCacheOptions::kNoProduceCache;
+  bool streaming_compile = false;
   SourceGroup* isolate_sources = nullptr;
   const char* icu_data_file = nullptr;
   const char* icu_locale = nullptr;
@@ -331,6 +332,9 @@ class Shell : public i::AllStatic {
   static bool EmptyMessageQueues(Isolate* isolate);
   static bool CompleteMessageLoop(Isolate* isolate);
 
+  static void PostForegroundTask(Isolate* isolate, std::unique_ptr<Task> task);
+  static void PostBlockingBackgroundTask(std::unique_ptr<Task> task);
+
   static std::unique_ptr<SerializationData> SerializeValue(
       Isolate* isolate, Local<Value> value, Local<Value> transfer);
   static MaybeLocal<Value> DeserializeValue(
@@ -446,6 +450,8 @@ class Shell : public i::AllStatic {
   static ArrayBuffer::Allocator* array_buffer_allocator;
 
   static void SetWaitUntilDone(Isolate* isolate, bool value);
+  static void NotifyStartStreamingTask(Isolate* isolate);
+  static void NotifyFinishStreamingTask(Isolate* isolate);
 
   static char* ReadCharsFromTcpPort(const char* name, int* size_out);
 
@@ -506,6 +512,7 @@ class Shell : public i::AllStatic {
   // the isolate_status_ needs to be concurrency-safe.
   static base::LazyMutex isolate_status_lock_;
   static std::map<Isolate*, bool> isolate_status_;
+  static std::map<Isolate*, int> isolate_running_streaming_tasks_;
 
   static base::LazyMutex cached_code_mutex_;
   static std::map<std::string, std::unique_ptr<ScriptCompiler::CachedData>>