Merged changes to avoid OOM for large wasm modules.

src/wasm/wasm-module.cc

 // Copyright 2015 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 <memory>
 
 #include "src/asmjs/asm-js.h"
 #include "src/assembler-inl.h"
 #include "src/base/atomic-utils.h"
+#include "src/base/utils/random-number-generator.h"
 #include "src/code-stubs.h"
 #include "src/compiler/wasm-compiler.h"
 #include "src/debug/interface-types.h"
 #include "src/frames-inl.h"
 #include "src/objects.h"
 #include "src/property-descriptor.h"
 #include "src/simulator.h"
 #include "src/snapshot/snapshot.h"
 #include "src/trap-handler/trap-handler.h"
 #include "src/v8.h"
@@ skipping 271 matching lines @@
 
 // A helper for compiling an entire module.
 class CompilationHelper {
  public:
   // The compilation helper takes ownership of the {WasmModule}.
   // In {CompileToModuleObject}, it will transfer ownership to the generated
   // {WasmModuleWrapper}. If this method is not called, ownership may be
   // reclaimed by explicitely releasing the {module_} field.
   CompilationHelper(Isolate* isolate, std::unique_ptr<WasmModule> module,
                     bool is_sync)
-      : isolate_(isolate), module_(std::move(module)), is_sync_(is_sync) {}
+      : isolate_(isolate),
+        module_(std::move(module)),
+        is_sync_(is_sync),
+        executed_units_(
+            isolate->random_number_generator(),
+            (isolate->heap()->memory_allocator()->code_range()->valid()
+                 ? isolate->heap()->memory_allocator()->code_range()->size()
+                 : isolate->heap()->code_space()->Capacity()) /
+                2),
+        num_background_tasks_(Min(
+            static_cast<size_t>(FLAG_wasm_num_compilation_tasks),
+            V8::GetCurrentPlatform()->NumberOfAvailableBackgroundThreads())),
+        stopped_compilation_tasks_(num_background_tasks_) {}
+
+  bool GetNextUncompiledFunctionId(size_t* index) {
+    DCHECK_NOT_NULL(index);
+    // - 1 because AtomicIncrement returns the value after the atomic increment.
+    *index = next_unit_.Increment(1) - 1;
+    return *index < compilation_units_.size();
+  }
 
   // The actual runnable task that performs compilations in the background.
   class CompilationTask : public CancelableTask {
    public:
     CompilationHelper* helper_;
     explicit CompilationTask(CompilationHelper* helper)
-        : CancelableTask(helper->isolate_), helper_(helper) {}
+        : CancelableTask(helper->isolate_, &helper->background_task_manager_),
+          helper_(helper) {}
 
     void RunInternal() override {
-      while (helper_->FetchAndExecuteCompilationUnit()) {
+      size_t index = 0;
+      while (helper_->executed_units_.CanAcceptWork() &&
+             helper_->GetNextUncompiledFunctionId(&index)) {
+        helper_->CompileAndSchedule(index);
       }
-      helper_->module_->pending_tasks.get()->Signal();
+      helper_->OnBackgroundTaskStopped();
     }
   };
 
+  void OnBackgroundTaskStopped() {
+    base::LockGuard<base::Mutex> guard(&tasks_mutex_);
+    ++stopped_compilation_tasks_;
+    DCHECK_LE(stopped_compilation_tasks_, num_background_tasks_);
+  }
+
+  void CompileAndSchedule(size_t index) {
+    DisallowHeapAllocation no_allocation;
+    DisallowHandleAllocation no_handles;
+    DisallowHandleDereference no_deref;
+    DisallowCodeDependencyChange no_dependency_change;
+    DCHECK_LT(index, compilation_units_.size());
+
+    std::unique_ptr<compiler::WasmCompilationUnit> unit =
+        std::move(compilation_units_.at(index));
+    unit->ExecuteCompilation();
+    {
+      base::LockGuard<base::Mutex> guard(&result_mutex_);
+      executed_units_.Schedule(std::move(unit));
+    }
+  }
+
+  class CodeGenerationSchedule {
+   public:
+    explicit CodeGenerationSchedule(
+        base::RandomNumberGenerator* random_number_generator,
+        size_t max_memory = 0);
+
+    void Schedule(std::unique_ptr<compiler::WasmCompilationUnit>&& item);
+
+    bool IsEmpty() const { return schedule_.empty(); }
+
+    std::unique_ptr<compiler::WasmCompilationUnit> GetNext();
+
+    bool CanAcceptWork() const;
+
+    void EnableThrottling() { throttle_ = true; }
+
+   private:
+    size_t GetRandomIndexInSchedule();
+
+    base::RandomNumberGenerator* random_number_generator_ = nullptr;
+    std::vector<std::unique_ptr<compiler::WasmCompilationUnit>> schedule_;
+    const size_t max_memory_;
+    bool throttle_ = false;
+    base::AtomicNumber<size_t> allocated_memory_{0};
+  };
+
   Isolate* isolate_;
   std::unique_ptr<WasmModule> module_;
   bool is_sync_;
   std::vector<std::unique_ptr<compiler::WasmCompilationUnit>>
       compilation_units_;
-  std::queue<std::unique_ptr<compiler::WasmCompilationUnit>> executed_units_;
+  CodeGenerationSchedule executed_units_;
   base::Mutex result_mutex_;
   base::AtomicNumber<size_t> next_unit_;
-  size_t num_background_tasks_ = 0;
+  const size_t num_background_tasks_ = 0;
+  CancelableTaskManager background_task_manager_;
 
   // Run by each compilation task and by the main thread.
   bool FetchAndExecuteCompilationUnit() {
     DisallowHeapAllocation no_allocation;
     DisallowHandleAllocation no_handles;
     DisallowHandleDereference no_deref;
     DisallowCodeDependencyChange no_dependency_change;
 
     // - 1 because AtomicIncrement returns the value after the atomic increment.
     size_t index = next_unit_.Increment(1) - 1;
     if (index >= compilation_units_.size()) {
       return false;
     }
 
     std::unique_ptr<compiler::WasmCompilationUnit> unit =
         std::move(compilation_units_.at(index));
     unit->ExecuteCompilation();
     base::LockGuard<base::Mutex> guard(&result_mutex_);
-    executed_units_.push(std::move(unit));
+    executed_units_.Schedule(std::move(unit));
     return true;
   }
 
   size_t InitializeParallelCompilation(
       const std::vector<WasmFunction>& functions, ModuleBytesEnv& module_env) {
     uint32_t start = module_env.module_env.module->num_imported_functions +
                      FLAG_skip_compiling_wasm_funcs;
     uint32_t num_funcs = static_cast<uint32_t>(functions.size());
     uint32_t funcs_to_compile = start > num_funcs ? 0 : num_funcs - start;
     compilation_units_.reserve(funcs_to_compile);
     for (uint32_t i = start; i < num_funcs; ++i) {
       const WasmFunction* func = &functions[i];
       constexpr bool is_sync = true;
       compilation_units_.push_back(
           std::unique_ptr<compiler::WasmCompilationUnit>(
               new compiler::WasmCompilationUnit(isolate_, &module_env, func,
                                                 !is_sync)));
     }
     return funcs_to_compile;
   }
 
-  void InitializeHandles() {
-    for (auto& unit : compilation_units_) {
-      unit->InitializeHandles();
+  void RestartCompilationTasks() {
+    base::LockGuard<base::Mutex> guard(&tasks_mutex_);
+    for (; stopped_compilation_tasks_ > 0; --stopped_compilation_tasks_) {
+      V8::GetCurrentPlatform()->CallOnBackgroundThread(
+          new CompilationTask(this), v8::Platform::kShortRunningTask);
     }
   }
 
-  uint32_t* StartCompilationTasks() {
-    num_background_tasks_ =
-        Min(static_cast<size_t>(FLAG_wasm_num_compilation_tasks),
-            V8::GetCurrentPlatform()->NumberOfAvailableBackgroundThreads());
-    uint32_t* task_ids = new uint32_t[num_background_tasks_];
-    for (size_t i = 0; i < num_background_tasks_; ++i) {
-      CompilationTask* task = new CompilationTask(this);
-      task_ids[i] = task->id();
-      V8::GetCurrentPlatform()->CallOnBackgroundThread(
-          task, v8::Platform::kShortRunningTask);
-    }
-    return task_ids;
-  }
-
   void WaitForCompilationTasks(uint32_t* task_ids) {
     for (size_t i = 0; i < num_background_tasks_; ++i) {
       // If the task has not started yet, then we abort it. Otherwise we wait
       // for it to finish.
       if (isolate_->cancelable_task_manager()->TryAbort(task_ids[i]) !=
           CancelableTaskManager::kTaskAborted) {
         module_->pending_tasks.get()->Wait();
       }
     }
   }
 
-  void FinishCompilationUnits(std::vector<Handle<Code>>& results,
-                              ErrorThrower* thrower) {
+  size_t FinishCompilationUnits(std::vector<Handle<Code>>& results,
+                                ErrorThrower* thrower) {
+    size_t finished = 0;
     while (true) {
       int func_index = -1;
       Handle<Code> result = FinishCompilationUnit(thrower, &func_index);
       if (func_index < 0) break;
       results[func_index] = result;
+      ++finished;
     }
+    RestartCompilationTasks();
+    return finished;
   }
 
   Handle<Code> FinishCompilationUnit(ErrorThrower* thrower, int* func_index) {
     std::unique_ptr<compiler::WasmCompilationUnit> unit;
     {
       base::LockGuard<base::Mutex> guard(&result_mutex_);
-      if (executed_units_.empty()) return Handle<Code>::null();
-      unit = std::move(executed_units_.front());
-      executed_units_.pop();
+      if (executed_units_.IsEmpty()) return Handle<Code>::null();
+      unit = executed_units_.GetNext();
     }
     *func_index = unit->func_index();
     Handle<Code> result = unit->FinishCompilation(thrower);
     return result;
   }
 
   void CompileInParallel(ModuleBytesEnv* module_env,
                          std::vector<Handle<Code>>& results,
                          ErrorThrower* thrower) {
     const WasmModule* module = module_env->module_env.module;
@@ skipping 15 matching lines @@
     //    main thread waits for all {CompilationTask} instances to finish.
     // 5) The main thread finishes the compilation.
 
     // Turn on the {CanonicalHandleScope} so that the background threads can
     // use the node cache.
     CanonicalHandleScope canonical(isolate_);
 
     // 1) The main thread allocates a compilation unit for each wasm function
     //    and stores them in the vector {compilation_units}.
     InitializeParallelCompilation(module->functions, *module_env);
-    InitializeHandles();
 
-    // Objects for the synchronization with the background threads.
-    base::AtomicNumber<size_t> next_unit(
-        static_cast<size_t>(FLAG_skip_compiling_wasm_funcs));
+    executed_units_.EnableThrottling();
 
     // 2) The main thread spawns {CompilationTask} instances which run on
     //    the background threads.
-    std::unique_ptr<uint32_t[]> task_ids(StartCompilationTasks());
+    RestartCompilationTasks();
 
-    // 3.a) The background threads and the main thread pick one compilation
-    //      unit at a time and execute the parallel phase of the compilation
-    //      unit. After finishing the execution of the parallel phase, the
-    //      result is enqueued in {executed_units}.
-    while (FetchAndExecuteCompilationUnit()) {
+    size_t finished_functions = 0;
+    while (finished_functions < compilation_units_.size()) {
+      // 3.a) The background threads and the main thread pick one compilation
+      //      unit at a time and execute the parallel phase of the compilation
+      //      unit. After finishing the execution of the parallel phase, the
+      //      result is enqueued in {executed_units}.
+      size_t index = 0;
+      if (GetNextUncompiledFunctionId(&index)) {
+        CompileAndSchedule(index);
+      }
       // 3.b) If {executed_units} contains a compilation unit, the main thread
       //      dequeues it and finishes the compilation unit. Compilation units
       //      are finished concurrently to the background threads to save
       //      memory.
-      FinishCompilationUnits(results, thrower);
+      finished_functions += FinishCompilationUnits(results, thrower);
     }
     // 4) After the parallel phase of all compilation units has started, the
-    //    main thread waits for all {CompilationTask} instances to finish.
-    WaitForCompilationTasks(task_ids.get());
-    // Finish the compilation of the remaining compilation units.
-    FinishCompilationUnits(results, thrower);
+    //    main thread waits for all {CompilationTask} instances to finish -
+    //    which happens once they all realize there's no next work item to
+    //    process.
+    background_task_manager_.CancelAndWait();
   }
 
   void CompileSequentially(ModuleBytesEnv* module_env,
                            std::vector<Handle<Code>>& results,
                            ErrorThrower* thrower) {
     DCHECK(!thrower->error());
 
     const WasmModule* module = module_env->module_env.module;
     for (uint32_t i = FLAG_skip_compiling_wasm_funcs;
          i < module->functions.size(); ++i) {
@@ skipping 181 matching lines @@
                                                          wasm_code, exp.index);
       int export_index =
           static_cast<int>(module->functions.size() + func_index);
       code_table->set(export_index, *wrapper_code);
       RecordStats(isolate_, *wrapper_code, is_sync_);
       func_index++;
     }
 
     return WasmModuleObject::New(isolate_, compiled_module);
   }
+  size_t stopped_compilation_tasks_ = 0;
+  base::Mutex tasks_mutex_;
 };
 
+CompilationHelper::CodeGenerationSchedule::CodeGenerationSchedule(
+    base::RandomNumberGenerator* random_number_generator, size_t max_memory)
+    : random_number_generator_(random_number_generator),
+      max_memory_(max_memory) {
+  DCHECK_NOT_NULL(random_number_generator_);
+  DCHECK_GT(max_memory_, 0);
+}
+
+void CompilationHelper::CodeGenerationSchedule::Schedule(
+    std::unique_ptr<compiler::WasmCompilationUnit>&& item) {
+  size_t cost = item->memory_cost();
+  schedule_.push_back(std::move(item));
+  allocated_memory_.Increment(cost);
+}
+
+bool CompilationHelper::CodeGenerationSchedule::CanAcceptWork() const {
+  return (!throttle_ || allocated_memory_.Value() <= max_memory_);
+}
+
+std::unique_ptr<compiler::WasmCompilationUnit>
+CompilationHelper::CodeGenerationSchedule::GetNext() {
+  DCHECK(!IsEmpty());
+  size_t index = GetRandomIndexInSchedule();
+  auto ret = std::move(schedule_[index]);
+  std::swap(schedule_[schedule_.size() - 1], schedule_[index]);
+  schedule_.pop_back();
+  allocated_memory_.Decrement(ret->memory_cost());
+  return ret;
+}
+
+size_t CompilationHelper::CodeGenerationSchedule::GetRandomIndexInSchedule() {
+  double factor = random_number_generator_->NextDouble();
+  size_t index = (size_t)(factor * schedule_.size());
+  DCHECK_GE(index, 0);
+  DCHECK_LT(index, schedule_.size());
+  return index;
+}
+
 static void MemoryInstanceFinalizer(Isolate* isolate,
                                     WasmInstanceObject* instance) {
   DisallowHeapAllocation no_gc;
   // If the memory object is destroyed, nothing needs to be done here.
   if (!instance->has_memory_object()) return;
   Handle<WasmInstanceWrapper> instance_wrapper =
       handle(instance->instance_wrapper());
   DCHECK(WasmInstanceWrapper::IsWasmInstanceWrapper(*instance_wrapper));
   DCHECK(instance_wrapper->has_instance());
   bool has_prev = instance_wrapper->has_previous();
@@ skipping 1950 matching lines @@
   std::unique_ptr<WasmInstance> temp_instance_ = nullptr;
   size_t outstanding_units_ = 0;
   size_t num_background_tasks_ = 0;
 
   void ReopenHandlesInDeferredScope() {
     DeferredHandleScope deferred(isolate_);
     function_tables_ = handle(*function_tables_, isolate_);
     signature_tables_ = handle(*signature_tables_, isolate_);
     code_table_ = handle(*code_table_, isolate_);
     temp_instance_->ReopenHandles(isolate_);
-    helper_->InitializeHandles();
+    for (auto& unit : helper_->compilation_units_) {
+      unit->ReopenCentryStub();
+    }
     deferred_handles_.push_back(deferred.Detach());
   }
 
   void AsyncCompileFailed(ErrorThrower& thrower) {
     RejectPromise(isolate_, context_, thrower, module_promise_);
     // The AsyncCompileJob is finished, we resolved the promise, we do not need
     // the data anymore. We can delete the AsyncCompileJob object.
     if (!FLAG_verify_predictable) delete this;
   }
 
@@ skipping 539 matching lines @@
   {
     wasm::WasmName name = Vector<const char>::cast(
         compiled_module->GetRawFunctionName(func_index));
     // Copy to std::string, because the underlying string object might move on
     // the heap.
     func_name.assign(name.start(), static_cast<size_t>(name.length()));
   }
   ErrorThrower thrower(isolate, "WasmLazyCompile");
   compiler::WasmCompilationUnit unit(isolate, &module_env, body,
                                      CStrVector(func_name.c_str()), func_index);
-  unit.InitializeHandles();
   unit.ExecuteCompilation();
   Handle<Code> code = unit.FinishCompilation(&thrower);
 
   // If there is a pending error, something really went wrong. The module was
   // verified before starting execution with lazy compilation.
   // This might be OOM, but then we cannot continue execution anyway.
   CHECK(!thrower.error());
 
   Handle<FixedArray> deopt_data = isolate->factory()->NewFixedArray(2, TENURED);
   Handle<WeakCell> weak_instance = isolate->factory()->NewWeakCell(instance);
@@ skipping 119 matching lines @@
                                      callee_compiled->instruction_start());
     }
     DCHECK_EQ(non_compiled_functions.size(), idx);
   }
 
   Code* ret =
       Code::cast(compiled_module->code_table()->get(func_to_return_idx));
   DCHECK_EQ(Code::WASM_FUNCTION, ret->kind());
   return handle(ret, isolate);
 }