[wasm] Introduce the SyncStreamingDecoder

This CL introduces the SyncStreamingDecoder to support
streaming compilation when --single-threaded is set. The
SyncStreamingDecoder buffers all bytes it receives over
{OnBytesReceived}, and compiles them synchronously upon {Finish}.

In addition to introducing SyncStreamingDecoder, this CL does
the following changes:
* Redirect streaming compilation to the new streaming decoder if
  --no-wasm-async-compilation is set. This flag is set if
  --single-threaded is set.
* Extend the test-streaming-compilation.cc tests to test also the new
  streaming decoder.

R=thibaudm@chromium.org

Bug: v8:10548
Change-Id: I807e291a6060067c9835de4adf82bcb00321d995
Reviewed-on: https://chromium-review.googlesource.com/c/v8/v8/+/2209053
Commit-Queue: Andreas Haas <ahaas@chromium.org>
Reviewed-by: Thibaud Michaud <thibaudm@chromium.org>
Cr-Commit-Position: refs/heads/master@{#67955}

BUILD.gn

@@ -3128,6 +3128,7 @@ v8_source_set("v8_base_without_compiler") {
     "src/wasm/streaming-decoder.cc",
     "src/wasm/streaming-decoder.h",
     "src/wasm/struct-types.h",
+    "src/wasm/sync-streaming-decoder.cc",
     "src/wasm/value-type.h",
     "src/wasm/wasm-arguments.h",
     "src/wasm/wasm-code-manager.cc",

src/wasm/streaming-decoder.cc

@@ -40,23 +40,9 @@ class V8_EXPORT_PRIVATE AsyncStreamingDecoder : public StreamingDecoder {
   // StreamingProcessor should not be called anymore.
   void NotifyCompilationEnded() override { Fail(); }
 
-  // Caching support.
-  // Sets the callback that is called after the module is fully compiled.
-  using ModuleCompiledCallback =
-      std::function<void(const std::shared_ptr<NativeModule>&)>;
-  void SetModuleCompiledCallback(ModuleCompiledCallback callback) override;
-  // Passes previously compiled module bytes from the embedder's cache.
-  bool SetCompiledModuleBytes(
-      Vector<const uint8_t> compiled_module_bytes) override;
-
   void NotifyNativeModuleCreated(
       const std::shared_ptr<NativeModule>& native_module) override;
 
-  Vector<const char> url() override { return VectorOf(url_); }
-  void SetUrl(Vector<const char> url) override {
-    url_.assign(url.begin(), url.length());
-  }
-
  private:
   // The SectionBuffer is the data object for the content of a single section.
   // It stores all bytes of the section (including section id and section
@@ -222,21 +208,15 @@ class V8_EXPORT_PRIVATE AsyncStreamingDecoder : public StreamingDecoder {
 
   uint32_t module_offset() const { return module_offset_; }
 
-  bool deserializing() const { return !compiled_module_bytes_.empty(); }
-
   std::unique_ptr<StreamingProcessor> processor_;
   std::unique_ptr<DecodingState> state_;
   std::vector<std::shared_ptr<SectionBuffer>> section_buffers_;
   bool code_section_processed_ = false;
   uint32_t module_offset_ = 0;
   size_t total_size_ = 0;
-  std::string url_;
 
-  // Caching support.
-  ModuleCompiledCallback module_compiled_callback_ = nullptr;
   // We need wire bytes in an array for deserializing cached modules.
   std::vector<uint8_t> wire_bytes_for_deserializing_;
-  Vector<const uint8_t> compiled_module_bytes_;
 
   DISALLOW_COPY_AND_ASSIGN(AsyncStreamingDecoder);
 };
@@ -322,18 +302,6 @@ void AsyncStreamingDecoder::Abort() {
   Fail();
 }
 
-void AsyncStreamingDecoder::SetModuleCompiledCallback(
-    ModuleCompiledCallback callback) {
-  DCHECK_NULL(module_compiled_callback_);
-  module_compiled_callback_ = callback;
-}
-
-bool AsyncStreamingDecoder::SetCompiledModuleBytes(
-    Vector<const uint8_t> compiled_module_bytes) {
-  compiled_module_bytes_ = compiled_module_bytes;
-  return true;
-}
-
 namespace {
 
 class TopTierCompiledCallback {

src/wasm/streaming-decoder.h

@@ -12,6 +12,7 @@
 #include "src/utils/vector.h"
 #include "src/wasm/compilation-environment.h"
 #include "src/wasm/wasm-constants.h"
+#include "src/wasm/wasm-engine.h"
 #include "src/wasm/wasm-result.h"
 
 namespace v8 {
@@ -83,18 +84,40 @@ class V8_EXPORT_PRIVATE StreamingDecoder {
   // Sets the callback that is called after the module is fully compiled.
   using ModuleCompiledCallback =
       std::function<void(const std::shared_ptr<NativeModule>&)>;
-  virtual void SetModuleCompiledCallback(ModuleCompiledCallback callback) = 0;
+
+  void SetModuleCompiledCallback(ModuleCompiledCallback callback) {
+    module_compiled_callback_ = callback;
+  }
+
   // Passes previously compiled module bytes from the embedder's cache.
-  virtual bool SetCompiledModuleBytes(
-      Vector<const uint8_t> compiled_module_bytes) = 0;
+  bool SetCompiledModuleBytes(Vector<const uint8_t> compiled_module_bytes) {
+    compiled_module_bytes_ = compiled_module_bytes;
+    return true;
+  }
 
   virtual void NotifyNativeModuleCreated(
       const std::shared_ptr<NativeModule>& native_module) = 0;
 
-  virtual Vector<const char> url() = 0;
-  virtual void SetUrl(Vector<const char> url) = 0;
+  Vector<const char> url() { return VectorOf(url_); }
+
+  void SetUrl(Vector<const char> url) {
+    url_.assign(url.begin(), url.length());
+  }
+
   static std::unique_ptr<StreamingDecoder> CreateAsyncStreamingDecoder(
       std::unique_ptr<StreamingProcessor> processor);
+
+  static std::unique_ptr<StreamingDecoder> CreateSyncStreamingDecoder(
+      Isolate* isolate, const WasmFeatures& enabled, Handle<Context> context,
+      const char* api_method_name_for_errors,
+      std::shared_ptr<CompilationResultResolver> resolver);
+
+ protected:
+  bool deserializing() const { return !compiled_module_bytes_.empty(); }
+
+  std::string url_;
+  ModuleCompiledCallback module_compiled_callback_;
+  Vector<const uint8_t> compiled_module_bytes_;
 };
 
 }  // namespace wasm

src/wasm/sync-streaming-decoder.cc

+// Copyright 2020 the V8 project authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "src/execution/isolate.h"
+#include "src/wasm/streaming-decoder.h"
+#include "src/wasm/wasm-engine.h"
+#include "src/wasm/wasm-objects-inl.h"
+#include "src/wasm/wasm-serialization.h"
+
+namespace v8 {
+namespace internal {
+namespace wasm {
+
+class V8_EXPORT_PRIVATE SyncStreamingDecoder : public StreamingDecoder {
+ public:
+  SyncStreamingDecoder(Isolate* isolate, const WasmFeatures& enabled,
+                       Handle<Context> context,
+                       const char* api_method_name_for_errors,
+                       std::shared_ptr<CompilationResultResolver> resolver)
+      : isolate_(isolate),
+        enabled_(enabled),
+        context_(context),
+        api_method_name_for_errors_(api_method_name_for_errors),
+        resolver_(resolver) {}
+
+  // The buffer passed into OnBytesReceived is owned by the caller.
+  void OnBytesReceived(Vector<const uint8_t> bytes) override {
+    buffer_.emplace_back(bytes.size());
+    CHECK_EQ(buffer_.back().size(), bytes.size());
+    std::memcpy(buffer_.back().data(), bytes.data(), bytes.size());
+    buffer_size_ += bytes.size();
+  }
+
+  void Finish() override {
+    // We copy all received chunks into one byte buffer.
+    auto bytes = std::make_unique<uint8_t[]>(buffer_size_);
+    uint8_t* destination = bytes.get();
+    for (auto& chunk : buffer_) {
+      std::memcpy(destination, chunk.data(), chunk.size());
+      destination += chunk.size();
+    }
+    CHECK_EQ(destination - bytes.get(), buffer_size_);
+
+    // Check if we can deserialize the module from cache.
+    if (deserializing()) {
+      HandleScope scope(isolate_);
+      SaveAndSwitchContext saved_context(isolate_, *context_);
+
+      MaybeHandle<WasmModuleObject> module_object = DeserializeNativeModule(
+          isolate_, compiled_module_bytes_,
+          Vector<const uint8_t>(bytes.get(), buffer_size_), url());
+
+      if (!module_object.is_null()) {
+        Handle<WasmModuleObject> module = module_object.ToHandleChecked();
+        resolver_->OnCompilationSucceeded(module);
+        return;
+      }
+    }
+
+    // Compile the received bytes synchronously.
+    ModuleWireBytes wire_bytes(bytes.get(), bytes.get() + buffer_size_);
+    ErrorThrower thrower(isolate_, api_method_name_for_errors_);
+    MaybeHandle<WasmModuleObject> module_object =
+        isolate_->wasm_engine()->SyncCompile(isolate_, enabled_, &thrower,
+                                             wire_bytes);
+    if (thrower.error()) {
+      resolver_->OnCompilationFailed(thrower.Reify());
+      return;
+    }
+    Handle<WasmModuleObject> module = module_object.ToHandleChecked();
+    if (module_compiled_callback_) {
+      module_compiled_callback_(module->shared_native_module());
+    }
+    resolver_->OnCompilationSucceeded(module);
+  }
+
+  void Abort() override {
+    // Abort is fully handled by the API, we only clear the buffer.
+    buffer_.clear();
+  }
+
+  void NotifyCompilationEnded() override { buffer_.clear(); }
+
+  void NotifyNativeModuleCreated(
+      const std::shared_ptr<NativeModule>&) override {
+    // This function is only called from the {AsyncCompileJob}.
+    UNREACHABLE();
+  }
+
+ private:
+  Isolate* isolate_;
+  const WasmFeatures enabled_;
+  Handle<Context> context_;
+  const char* api_method_name_for_errors_;
+  std::shared_ptr<CompilationResultResolver> resolver_;
+
+  std::vector<std::vector<uint8_t>> buffer_;
+  size_t buffer_size_ = 0;
+};
+
+std::unique_ptr<StreamingDecoder> StreamingDecoder::CreateSyncStreamingDecoder(
+    Isolate* isolate, const WasmFeatures& enabled, Handle<Context> context,
+    const char* api_method_name_for_errors,
+    std::shared_ptr<CompilationResultResolver> resolver) {
+  return std::make_unique<SyncStreamingDecoder>(isolate, enabled, context,
+                                                api_method_name_for_errors,
+                                                std::move(resolver));
+}
+}  // namespace wasm
+}  // namespace internal
+}  // namespace v8

src/wasm/wasm-engine.cc

@@ -600,10 +600,14 @@ std::shared_ptr<StreamingDecoder> WasmEngine::StartStreamingCompilation(
     Isolate* isolate, const WasmFeatures& enabled, Handle<Context> context,
     const char* api_method_name,
     std::shared_ptr<CompilationResultResolver> resolver) {
-  AsyncCompileJob* job =
-      CreateAsyncCompileJob(isolate, enabled, std::unique_ptr<byte[]>(nullptr),
-                            0, context, api_method_name, std::move(resolver));
-  return job->CreateStreamingDecoder();
+  if (FLAG_wasm_async_compilation) {
+    AsyncCompileJob* job = CreateAsyncCompileJob(
+        isolate, enabled, std::unique_ptr<byte[]>(nullptr), 0, context,
+        api_method_name, std::move(resolver));
+    return job->CreateStreamingDecoder();
+  }
+  return StreamingDecoder::CreateSyncStreamingDecoder(
+      isolate, enabled, context, api_method_name, std::move(resolver));
 }
 
 void WasmEngine::CompileFunction(Isolate* isolate, NativeModule* native_module,

src/wasm/wasm-js.cc

@@ -2072,20 +2072,8 @@ void WasmJs::Install(Isolate* isolate, bool exposed_on_global_object) {
   InstallFunc(isolate, webassembly, "validate", WebAssemblyValidate, 1);
   InstallFunc(isolate, webassembly, "instantiate", WebAssemblyInstantiate, 1);
 
-  // The implementation of streaming compilation depends on async compilation.
-  // If async compilation is disabled, then a streaming compilation callback
-  // should not be set.
-  CHECK_IMPLIES(!FLAG_wasm_async_compilation,
-                isolate->wasm_streaming_callback() == nullptr);
-
   if (FLAG_wasm_test_streaming) {
-    if (FLAG_wasm_async_compilation) {
-      isolate->set_wasm_streaming_callback(WasmStreamingCallbackForTesting);
-    } else {
-      printf(
-          "--wasm-test-streaming gets ignored because async compilation is "
-          "disabled.");
-    }
+    isolate->set_wasm_streaming_callback(WasmStreamingCallbackForTesting);
   }
 
   if (isolate->wasm_streaming_callback() != nullptr) {

test/cctest/wasm/test-streaming-compilation.cc

@@ -7,7 +7,6 @@
 #include "src/objects/managed.h"
 #include "src/objects/objects-inl.h"
 #include "src/utils/vector.h"
-
 #include "src/wasm/module-decoder.h"
 #include "src/wasm/streaming-decoder.h"
 #include "src/wasm/wasm-engine.h"
@@ -16,9 +15,8 @@
 #include "src/wasm/wasm-objects-inl.h"
 #include "src/wasm/wasm-objects.h"
 #include "src/wasm/wasm-serialization.h"
-
 #include "test/cctest/cctest.h"
-
+#include "test/common/wasm/flag-utils.h"
 #include "test/common/wasm/test-signatures.h"
 #include "test/common/wasm/wasm-macro-gen.h"
 
@@ -174,13 +172,20 @@ class StreamTester {
 };
 }  // namespace
 
-#define STREAM_TEST(name)                               \
-  void RunStream_##name();                              \
-  TEST(name) {                                          \
-    MockPlatform platform;                              \
-    CcTest::InitializeVM();                             \
-    RunStream_##name();                                 \
-  }                                                     \
+#define STREAM_TEST(name)                                                     \
+  void RunStream_##name();                                                    \
+  TEST(Async##name) {                                                         \
+    MockPlatform platform;                                                    \
+    CcTest::InitializeVM();                                                   \
+    RunStream_##name();                                                       \
+  }                                                                           \
+                                                                              \
+  TEST(SingleThreaded##name) {                                                \
+    i::FlagScope<bool> single_threaded_scope(&i::FLAG_single_threaded, true); \
+    MockPlatform platform;                                                    \
+    CcTest::InitializeVM();                                                   \
+    RunStream_##name();                                                       \
+  }                                                                           \
   void RunStream_##name()
 
 // Create a valid module with 3 functions.