Introducing an event loop mechanism for d8.

This mechanism ensures APIs like wasm async complete their work, 
without requiring use of natives (%APIs).

The mechanism is similar to the one used in content_shell,
which should allow us to easily port tests in that environment.

Review-Url: https://codereview.chromium.org/2842843005
Cr-Original-Commit-Position: refs/heads/master@{#44908}
Bug: 
Change-Id: I9deee0d256a600c60b42902fc8ef8478e5546344
Reviewed-on: https://chromium-review.googlesource.com/494968
Commit-Queue: Mircea Trofin <mtrofin@google.com>
Reviewed-by: Jochen Eisinger <jochen@chromium.org>
Cr-Commit-Position: refs/heads/master@{#45165}

include/libplatform/libplatform.h

@@ -15,6 +15,11 @@ namespace platform {
 enum class IdleTaskSupport { kDisabled, kEnabled };
 enum class InProcessStackDumping { kDisabled, kEnabled };
 
+enum class MessageLoopBehavior : bool {
+  kDoNotWait = false,
+  kWaitForWork = true
+};
+
 /**
  * Returns a new instance of the default v8::Platform implementation.
  *
@@ -36,12 +41,16 @@ V8_PLATFORM_EXPORT v8::Platform* CreateDefaultPlatform(
  * Pumps the message loop for the given isolate.
  *
  * The caller has to make sure that this is called from the right thread.
- * Returns true if a task was executed, and false otherwise. This call does
- * not block if no task is pending. The |platform| has to be created using
- * |CreateDefaultPlatform|.
+ * Returns true if a task was executed, and false otherwise. Unless requested
+ * through the |behavior| parameter, this call does not block if no task is
+ * pending. The |platform| has to be created using |CreateDefaultPlatform|.
  */
-V8_PLATFORM_EXPORT bool PumpMessageLoop(v8::Platform* platform,
-                                        v8::Isolate* isolate);
+V8_PLATFORM_EXPORT bool PumpMessageLoop(
+    v8::Platform* platform, v8::Isolate* isolate,
+    MessageLoopBehavior behavior = MessageLoopBehavior::kDoNotWait);
+
+V8_PLATFORM_EXPORT void EnsureEventLoopInitialized(v8::Platform* platform,
+                                                   v8::Isolate* isolate);
 
 /**
  * Runs pending idle tasks for at most |idle_time_in_seconds| seconds.

src/d8.cc

@@ -422,6 +422,7 @@ base::LazyMutex Shell::workers_mutex_;
 bool Shell::allow_new_workers_ = true;
 i::List<Worker*> Shell::workers_;
 std::vector<ExternalizedContents> Shell::externalized_contents_;
+std::map<v8::Isolate*, bool> Shell::isolate_status;
 
 Global<Context> Shell::evaluation_context_;
 ArrayBuffer::Allocator* Shell::array_buffer_allocator;
@@ -1345,6 +1346,18 @@ void Shell::Quit(const v8::FunctionCallbackInfo<v8::Value>& args) {
                  const_cast<v8::FunctionCallbackInfo<v8::Value>*>(&args));
 }
 
+void Shell::WaitUntilDone(const v8::FunctionCallbackInfo<v8::Value>& args) {
+  if (isolate_status.count(args.GetIsolate()) == 0) {
+    isolate_status.insert(std::make_pair(args.GetIsolate(), true));
+  } else {
+    isolate_status[args.GetIsolate()] = true;
+  }
+}
+
+void Shell::NotifyDone(const v8::FunctionCallbackInfo<v8::Value>& args) {
+  DCHECK_EQ(isolate_status.count(args.GetIsolate()), 1);
+  isolate_status[args.GetIsolate()] = false;
+}
 
 void Shell::Version(const v8::FunctionCallbackInfo<v8::Value>& args) {
   args.GetReturnValue().Set(
@@ -1582,6 +1595,19 @@ Local<ObjectTemplate> Shell::CreateGlobalTemplate(Isolate* isolate) {
             .ToLocalChecked(),
         FunctionTemplate::New(isolate, Quit));
   }
+  Local<ObjectTemplate> test_template = ObjectTemplate::New(isolate);
+  global_template->Set(
+      String::NewFromUtf8(isolate, "testRunner", NewStringType::kNormal)
+          .ToLocalChecked(),
+      test_template);
+  test_template->Set(
+      String::NewFromUtf8(isolate, "notifyDone", NewStringType::kNormal)
+          .ToLocalChecked(),
+      FunctionTemplate::New(isolate, NotifyDone));
+  test_template->Set(
+      String::NewFromUtf8(isolate, "waitUntilDone", NewStringType::kNormal)
+          .ToLocalChecked(),
+      FunctionTemplate::New(isolate, WaitUntilDone));
   global_template->Set(
       String::NewFromUtf8(isolate, "version", NewStringType::kNormal)
           .ToLocalChecked(),
@@ -2266,6 +2292,8 @@ void SourceGroup::ExecuteInThread() {
   create_params.host_import_module_dynamically_callback_ =
       Shell::HostImportModuleDynamically;
   Isolate* isolate = Isolate::New(create_params);
+
+  v8::platform::EnsureEventLoopInitialized(g_platform, isolate);
   D8Console console(isolate);
   debug::SetConsoleDelegate(isolate, &console);
   for (int i = 0; i < Shell::options.stress_runs; ++i) {
@@ -2290,6 +2318,8 @@ void SourceGroup::ExecuteInThread() {
     done_semaphore_.Signal();
   }
 
+  Shell::CompleteMessageLoop(isolate);
+
   isolate->Dispose();
 }
 
@@ -2646,6 +2676,7 @@ int Shell::RunMain(Isolate* isolate, int argc, char* argv[], bool last_run) {
     options.isolate_sources[i].StartExecuteInThread();
   }
   {
+    v8::platform::EnsureEventLoopInitialized(g_platform, isolate);
     if (options.lcov_file) {
       debug::Coverage::SelectMode(isolate, debug::Coverage::kPreciseCount);
     }
@@ -2668,6 +2699,7 @@ int Shell::RunMain(Isolate* isolate, int argc, char* argv[], bool last_run) {
     WriteLcovData(isolate, options.lcov_file);
   }
   CollectGarbage(isolate);
+  CompleteMessageLoop(isolate);
   for (int i = 1; i < options.num_isolates; ++i) {
     if (last_run) {
       options.isolate_sources[i].JoinThread();
@@ -2695,24 +2727,28 @@ void Shell::CollectGarbage(Isolate* isolate) {
   }
 }
 
+void Shell::CompleteMessageLoop(Isolate* isolate) {
+  while (v8::platform::PumpMessageLoop(
+      g_platform, isolate,
+      Shell::isolate_status[isolate]
+          ? platform::MessageLoopBehavior::kWaitForWork
+          : platform::MessageLoopBehavior::kDoNotWait)) {
+    isolate->RunMicrotasks();
+  }
+  v8::platform::RunIdleTasks(g_platform, isolate,
+                             50.0 / base::Time::kMillisecondsPerSecond);
+}
+
 void Shell::EmptyMessageQueues(Isolate* isolate) {
   if (i::FLAG_verify_predictable) return;
-  while (true) {
-    // Pump the message loop until it is empty.
-    while (v8::platform::PumpMessageLoop(g_platform, isolate)) {
-      isolate->RunMicrotasks();
-    }
-    // Run the idle tasks.
-    v8::platform::RunIdleTasks(g_platform, isolate,
-                               50.0 / base::Time::kMillisecondsPerSecond);
-    // If there are still outstanding waiters, sleep a little (to wait for
-    // background tasks) and then try everything again.
-    if (reinterpret_cast<i::Isolate*>(isolate)->GetWaitCountForTesting() > 0) {
-      base::OS::Sleep(base::TimeDelta::FromMilliseconds(1));
-    } else {
-      break;
-    }
+  // Pump the message loop until it is empty.
+  while (v8::platform::PumpMessageLoop(
+      g_platform, isolate, platform::MessageLoopBehavior::kDoNotWait)) {
+    isolate->RunMicrotasks();
   }
+  // Run the idle tasks.
+  v8::platform::RunIdleTasks(g_platform, isolate,
+                             50.0 / base::Time::kMillisecondsPerSecond);
 }
 
 class Serializer : public ValueSerializer::Delegate {

src/d8.h

@@ -6,6 +6,7 @@
 #define V8_D8_H_
 
 #include <iterator>
+#include <map>
 #include <memory>
 #include <string>
 #include <vector>
@@ -356,6 +357,7 @@ class Shell : public i::AllStatic {
   static void OnExit(Isolate* isolate);
   static void CollectGarbage(Isolate* isolate);
   static void EmptyMessageQueues(Isolate* isolate);
+  static void CompleteMessageLoop(Isolate* isolate);
 
   static std::unique_ptr<SerializationData> SerializeValue(
       Isolate* isolate, Local<Value> value, Local<Value> transfer);
@@ -391,6 +393,8 @@ class Shell : public i::AllStatic {
   static void Print(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void PrintErr(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Write(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void WaitUntilDone(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void NotifyDone(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void QuitOnce(v8::FunctionCallbackInfo<v8::Value>* args);
   static void Quit(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Version(const v8::FunctionCallbackInfo<v8::Value>& args);
@@ -454,6 +458,7 @@ class Shell : public i::AllStatic {
   static const char* kPrompt;
   static ShellOptions options;
   static ArrayBuffer::Allocator* array_buffer_allocator;
+  static std::map<Isolate*, bool> isolate_status;
 
  private:
   static Global<Context> evaluation_context_;

src/isolate.h

@@ -1291,11 +1291,6 @@ class Isolate {
   // reset to nullptr.
   void UnregisterFromReleaseAtTeardown(ManagedObjectFinalizer** finalizer_ptr);
 
-  // Used by mjsunit tests to force d8 to wait for certain things to run.
-  inline void IncrementWaitCountForTesting() { wait_count_++; }
-  inline void DecrementWaitCountForTesting() { wait_count_--; }
-  inline int GetWaitCountForTesting() { return wait_count_; }
-
  protected:
   explicit Isolate(bool enable_serializer);
   bool IsArrayOrObjectPrototype(Object* object);
@@ -1582,8 +1577,6 @@ class Isolate {
 
   size_t total_regexp_code_generated_;
 
-  int wait_count_ = 0;
-
   friend class ExecutionAccess;
   friend class HandleScopeImplementer;
   friend class HeapTester;

src/libplatform/default-platform.cc

@@ -41,9 +41,15 @@ v8::Platform* CreateDefaultPlatform(
   return platform;
 }
 
+bool PumpMessageLoop(v8::Platform* platform, v8::Isolate* isolate,
+                     MessageLoopBehavior behavior) {
+  return reinterpret_cast<DefaultPlatform*>(platform)->PumpMessageLoop(
+      isolate, behavior);
+}
 
-bool PumpMessageLoop(v8::Platform* platform, v8::Isolate* isolate) {
-  return reinterpret_cast<DefaultPlatform*>(platform)->PumpMessageLoop(isolate);
+void EnsureEventLoopInitialized(v8::Platform* platform, v8::Isolate* isolate) {
+  return reinterpret_cast<DefaultPlatform*>(platform)
+      ->EnsureEventLoopInitialized(isolate);
 }
 
 void RunIdleTasks(v8::Platform* platform, v8::Isolate* isolate,
@@ -158,7 +164,23 @@ IdleTask* DefaultPlatform::PopTaskInMainThreadIdleQueue(v8::Isolate* isolate) {
   return task;
 }
 
-bool DefaultPlatform::PumpMessageLoop(v8::Isolate* isolate) {
+void DefaultPlatform::EnsureEventLoopInitialized(v8::Isolate* isolate) {
+  if (event_loop_control_.count(isolate) == 0) {
+    event_loop_control_.insert(std::make_pair(
+        isolate, std::unique_ptr<base::Semaphore>(new base::Semaphore(0))));
+  }
+}
+
+void DefaultPlatform::WaitForForegroundWork(v8::Isolate* isolate) {
+  DCHECK_EQ(event_loop_control_.count(isolate), 1);
+  event_loop_control_[isolate]->Wait();
+}
+
+bool DefaultPlatform::PumpMessageLoop(v8::Isolate* isolate,
+                                      MessageLoopBehavior behavior) {
+  if (behavior == MessageLoopBehavior::kWaitForWork) {
+    WaitForForegroundWork(isolate);
+  }
   Task* task = NULL;
   {
     base::LockGuard<base::Mutex> guard(&lock_);
@@ -166,14 +188,14 @@ bool DefaultPlatform::PumpMessageLoop(v8::Isolate* isolate) {
     // Move delayed tasks that hit their deadline to the main queue.
     task = PopTaskInMainThreadDelayedQueue(isolate);
     while (task != NULL) {
-      main_thread_queue_[isolate].push(task);
+      ScheduleOnForegroundThread(isolate, task);
       task = PopTaskInMainThreadDelayedQueue(isolate);
     }
 
     task = PopTaskInMainThreadQueue(isolate);
 
     if (task == NULL) {
-      return false;
+      return behavior == MessageLoopBehavior::kWaitForWork;
     }
   }
   task->Run();
@@ -206,10 +228,17 @@ void DefaultPlatform::CallOnBackgroundThread(Task* task,
   queue_.Append(task);
 }
 
+void DefaultPlatform::ScheduleOnForegroundThread(v8::Isolate* isolate,
+                                                 Task* task) {
+  main_thread_queue_[isolate].push(task);
+  if (event_loop_control_.count(isolate) != 0) {
+    event_loop_control_[isolate]->Signal();
+  }
+}
 
 void DefaultPlatform::CallOnForegroundThread(v8::Isolate* isolate, Task* task) {
   base::LockGuard<base::Mutex> guard(&lock_);
-  main_thread_queue_[isolate].push(task);
+  ScheduleOnForegroundThread(isolate, task);
 }
 
 

src/libplatform/default-platform.h

@@ -41,7 +41,10 @@ class V8_PLATFORM_EXPORT DefaultPlatform : public NON_EXPORTED_BASE(Platform) {
 
   void EnsureInitialized();
 
-  bool PumpMessageLoop(v8::Isolate* isolate);
+  bool PumpMessageLoop(
+      v8::Isolate* isolate,
+      MessageLoopBehavior behavior = MessageLoopBehavior::kDoNotWait);
+  void EnsureEventLoopInitialized(v8::Isolate* isolate);
 
   void RunIdleTasks(v8::Isolate* isolate, double idle_time_in_seconds);
 
@@ -81,6 +84,9 @@ class V8_PLATFORM_EXPORT DefaultPlatform : public NON_EXPORTED_BASE(Platform) {
   Task* PopTaskInMainThreadDelayedQueue(v8::Isolate* isolate);
   IdleTask* PopTaskInMainThreadIdleQueue(v8::Isolate* isolate);
 
+  void WaitForForegroundWork(v8::Isolate* isolate);
+  void ScheduleOnForegroundThread(v8::Isolate* isolate, Task* task);
+
   base::Mutex lock_;
   bool initialized_;
   int thread_pool_size_;
@@ -89,6 +95,7 @@ class V8_PLATFORM_EXPORT DefaultPlatform : public NON_EXPORTED_BASE(Platform) {
   TaskQueue queue_;
   std::map<v8::Isolate*, std::queue<Task*>> main_thread_queue_;
   std::map<v8::Isolate*, std::queue<IdleTask*>> main_thread_idle_queue_;
+  std::map<v8::Isolate*, std::unique_ptr<base::Semaphore>> event_loop_control_;
 
   typedef std::pair<double, Task*> DelayedEntry;
   std::map<v8::Isolate*,

src/runtime/runtime-test.cc

@@ -1019,15 +1019,5 @@ RUNTIME_FUNCTION(Runtime_RedirectToWasmInterpreter) {
   return isolate->heap()->undefined_value();
 }
 
-RUNTIME_FUNCTION(Runtime_IncrementWaitCount) {
-  isolate->IncrementWaitCountForTesting();
-  return isolate->heap()->undefined_value();
-}
-
-RUNTIME_FUNCTION(Runtime_DecrementWaitCount) {
-  isolate->DecrementWaitCountForTesting();
-  return isolate->heap()->undefined_value();
-}
-
 }  // namespace internal
 }  // namespace v8

src/runtime/runtime.h

@@ -471,9 +471,7 @@ namespace internal {
   F(PromiseRevokeReject, 1, 1)              \
   F(PromiseResult, 1, 1)                    \
   F(PromiseStatus, 1, 1)                    \
-  F(ReportPromiseReject, 2, 1)              \
-  F(IncrementWaitCount, 0, 1)               \
-  F(DecrementWaitCount, 0, 1)
+  F(ReportPromiseReject, 2, 1)
 
 #define FOR_EACH_INTRINSIC_PROXY(F)     \
   F(IsJSProxy, 1, 1)                    \

test/debugger/debug/debug-stepin-accessor.js

@@ -233,7 +233,8 @@ function testProtoSetter1_2() {
 }
 
 for (var n in this) {
-  if (n.substr(0, 4) != 'test') {
+  if (n.substr(0, 4) != 'test' ||
+     n == 'testRunner') {
     continue;
   }
   state = 1;

test/mjsunit/basic-promise.js

@@ -8,14 +8,6 @@
 // exceptions which are swallowed in a then clause.
 failWithMessage = (msg) => %AbortJS(msg);
 
-let decrement = () => { %DecrementWaitCount(); }
-let increment = () => { %IncrementWaitCount(); }
-
-function WaitForPromise(p) {
-  increment();
-  p.then(decrement, decrement);
-}
-
 function newPromise() {
   var outerResolve;
   var outerReject;
@@ -23,7 +15,7 @@ function newPromise() {
     outerResolve = resolve;
     outerReject = reject;
   });
-  WaitForPromise(promise);  // explicitly wait for promise to resolve.
+  Promise.resolve(promise);
   return {
     resolve: outerResolve,
     reject: outerReject,

test/mjsunit/mjsunit.js

@@ -123,6 +123,9 @@ var assertMatches;
 // Assert the result of a promise.
 var assertPromiseResult;
 
+var promiseTestChain;
+var promiseTestCount = 0;
+
 // These bits must be in sync with bits defined in Runtime_GetOptimizationStatus
 var V8OptimizationStatus = {
   kIsFunction: 1 << 0,
@@ -499,21 +502,35 @@ var failWithMessage;
     // We have to patch mjsunit because normal assertion failures just throw
     // exceptions which are swallowed in a then clause.
     // We use eval here to avoid parsing issues with the natives syntax.
-    failWithMessage = (msg) => eval("%AbortJS(msg)");
-    if (!fail)
-      fail = result => failWithMessage("assertPromiseResult failed: " + result);
+    if (!success) success = () => {};
 
-    eval("%IncrementWaitCount()");
-    return promise.then(
-      result => {
-        eval("%DecrementWaitCount()");
-        success(result);
-      },
-      result => {
-        eval("%DecrementWaitCount()");
-        fail(result);
-      }
-    );
+    failWithMessage = (msg) => eval("%AbortJS(msg)");
+    if (!fail) {
+      fail = result => failWithMessage("assertPromiseResult failed: " + result);
+    }
+
+    var test_promise =
+        promise.then(
+          result => {
+            try {
+              success(result);
+            } catch (e) {
+              fail(e);
+            }
+          },
+          result => {
+            fail(result);
+          }
+        )
+        .then((x)=> {
+          if (--promiseTestCount == 0) testRunner.notifyDone();
+        });
+
+    if (!promiseTestChain) promiseTestChain = Promise.resolve();
+    // waitUntilDone is idempotent.
+    testRunner.waitUntilDone();
+    ++promiseTestCount;
+    return promiseTestChain.then(test_promise);
   };
 
   var OptimizationStatusImpl = undefined;

test/mjsunit/wasm/async-compile.js

@@ -20,14 +20,7 @@ function assertCompileError(buffer) {
       ex => assertTrue(ex instanceof WebAssembly.CompileError));
 }
 
-// These tests execute asynchronously. In order to avoid executing several tests
-// concurrently (which makes debugging much harder), build a promise chain to
-// start the next task only after the previous one ended.
-
-let testChain = Promise.resolve();
-let addTest = fun => testChain = testChain.then(() => fun());
-
-addTest(async function basicCompile() {
+assertPromiseResult(async function basicCompile() {
   let ok_buffer = (() => {
     var builder = new WasmModuleBuilder();
     builder.addFunction('f', kSig_i_v)
@@ -58,9 +51,9 @@ addTest(async function basicCompile() {
   for (var i = 0; i < kNumCompiles; i++) {
     await assertCompileError(bad_buffer);
   }
-});
+}());
 
-addTest(async function badFunctionInTheMiddle() {
+assertPromiseResult(async function badFunctionInTheMiddle() {
   // We had an error where an exception was generated by a background task and
   // later thrown in a foreground task. The handle to the exception died
   // inbetween, since the HandleScope was left.
@@ -76,4 +69,4 @@ addTest(async function badFunctionInTheMiddle() {
   }
   let buffer = builder.toBuffer();
   await assertCompileError(buffer);
-});
+}());

test/mjsunit/wasm/compilation-limits.js

@@ -106,8 +106,4 @@ async function TestAll() {
   await FailAsyncInstantiate();
 }
 
-%IncrementWaitCount();
-TestAll().then(
-  () => { %DecrementWaitCount(); },
-  () => { %DecrementWaitCount(); }
-);
+assertPromiseResult(TestAll());

test/mjsunit/wasm/instantiate-module-basic.js

@@ -70,7 +70,7 @@ function CheckInstance(instance) {
 
   print('async instantiate...');
   let instance_promise = WebAssembly.instantiate(buffer);
-  assertPromiseResult(instance_promise, CheckInstance);
+  assertPromiseResult(instance_promise, pair => CheckInstance(pair.instance));
 })();
 
 // Check that validate works correctly for a module.

test/mjsunit/wasm/wasm-api-overloading.js

@@ -30,8 +30,6 @@ assertPromiseResult(
     return WebAssembly.instantiate(wrapper);
   })(),
   pair => {
-    print(2);
-    var pair = result.pair;
     assertTrue(pair.instance instanceof WebAssembly.Instance);
     assertTrue(pair.module instanceof WebAssembly.Module);
     %ResetWasmOverloads();