Add support for flushing microtasks in dart_engine.h

Also add a sample for calling Dart functions which return/accept futures.

Change-Id: I932d0577709f4e068ccd3212797751ecb347c4f9
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/408281
Commit-Queue: Ivan Inozemtsev <[email protected]>
Reviewed-by: Martin Kustermann <[email protected]>

Author: iinozemtsev

runtime/engine/dart_engine_impl.cc

@@ -72,5 +72,9 @@ DART_EXPORT void DartEngine_HandleMessage(Dart_Isolate isolate) {
   Engine::instance()->HandleMessage(isolate);
 }
 
+DART_EXPORT Dart_Handle DartEngine_DrainMicrotasksQueue() {
+  return Engine::instance()->DrainMicrotasksQueue();
+}
+
 }  // namespace engine
 }  // namespace dart

runtime/engine/engine.cc

@@ -4,7 +4,6 @@
 
 #include "engine/engine.h"
 #include <memory>
-#include <utility>
 #include "bin/dartutils.h"
 #include "include/dart_api.h"
 #include "include/dart_embedder_api.h"
@@ -16,6 +15,8 @@
 namespace dart {
 namespace engine {
 
+constexpr char kRunPendingImmediateCallback[] = "_runPendingImmediateCallback";
+
 using platform::MutexLocker;
 
 Engine* Engine::instance() {
@@ -195,8 +196,8 @@ Dart_Isolate Engine::StartIsolate(DartEngine_SnapshotData snapshot,
   Dart_Handle core_libs_result =
       bin::DartUtils::PrepareForScriptLoading(false, false);
   if (Dart_IsError(core_libs_result)) {
-    Dart_ShutdownIsolate();
     *error = Utils::StrDup(Dart_GetError(core_libs_result));
+    Dart_ShutdownIsolate();
     return nullptr;
   }
 
@@ -210,12 +211,27 @@ Dart_Isolate Engine::StartIsolate(DartEngine_SnapshotData snapshot,
         snapshot.kernel_buffer, snapshot.kernel_buffer_size);
 
     if (Dart_IsError(library)) {
-      Dart_ShutdownIsolate();
       *error = Utils::StrDup(Dart_GetError(library));
+      Dart_ShutdownIsolate();
       return nullptr;
     }
   }
 
+  Dart_Handle isolate_library = Dart_LookupLibrary(
+      Dart_NewStringFromCString(bin::DartUtils::kIsolateLibURL));
+  if (Dart_IsError(isolate_library)) {
+    *error = Utils::StrDup(Dart_GetError(isolate_library));
+    Dart_ShutdownIsolate();
+    return nullptr;
+  }
+
+  std::shared_ptr<Engine::IsolateData> isolate_data = DataForIsolate(isolate);
+  isolate_data->isolate_library = Dart_NewPersistentHandle(isolate_library);
+  isolate_data->drain_microtasks_function_name = Dart_NewPersistentHandle(
+      Dart_NewStringFromCString(kRunPendingImmediateCallback));
+  isolate_data->scheduler.context = nullptr;
+  isolate_data->scheduler.schedule_callback = nullptr;
+
   Dart_ExitScope();
   Dart_ExitIsolate();
   is_running_ = true;
@@ -232,9 +248,12 @@ void Engine::Shutdown() {
 
   is_running_ = false;
   for (auto isolate : isolates_) {
+    std::shared_ptr<Engine::IsolateData> isolate_data = DataForIsolate(isolate);
     LockIsolate(isolate);
     Dart_EnterIsolate(isolate);
     Dart_SetMessageNotifyCallback(nullptr);
+    Dart_DeletePersistentHandle(isolate_data->isolate_library);
+    Dart_DeletePersistentHandle(isolate_data->drain_microtasks_function_name);
     Dart_ShutdownIsolate();
     UnlockIsolate(isolate);
   }
@@ -277,17 +296,30 @@ void Engine::HandleMessage(Dart_Isolate isolate) {
   UnlockIsolate(isolate);
 }
 
-Mutex& Engine::MutexForIsolate(Dart_Isolate isolate) {
+Dart_Handle Engine::DrainMicrotasksQueue() {
+  std::shared_ptr<Engine::IsolateData> isolate_data =
+      DataForIsolate(Dart_CurrentIsolate());
+  return Dart_Invoke(isolate_data->isolate_library,
+                     isolate_data->drain_microtasks_function_name, 0, nullptr);
+}
+
+std::shared_ptr<Engine::IsolateData> Engine::DataForIsolate(
+    Dart_Isolate isolate) {
   MutexLocker ml(&engine_state_);
-  return mutexes_[isolate];
+  auto it = isolate_data_.find(isolate);
+  if (it == isolate_data_.end()) {
+    it = isolate_data_.emplace(isolate, std::make_shared<Engine::IsolateData>())
+             .first;
+  }
+  return it->second;
 }
 
 void Engine::LockIsolate(Dart_Isolate isolate) {
-  MutexForIsolate(isolate).Lock();
+  DataForIsolate(isolate)->mutex.Lock();
 }
 
 void Engine::UnlockIsolate(Dart_Isolate isolate) {
-  MutexForIsolate(isolate).Unlock();
+  DataForIsolate(isolate)->mutex.Unlock();
 }
 
 void Engine::NotifyMessage(Dart_Isolate isolate) {
@@ -313,11 +345,7 @@ void Engine::NotifyMessage(Dart_Isolate isolate) {
     return;
   }
 
-  DartEngine_MessageScheduler scheduler;
-  {
-    MutexLocker ml(&engine_state_);
-    scheduler = schedulers_[isolate];
-  }
+  DartEngine_MessageScheduler scheduler = DataForIsolate(isolate)->scheduler;
 
   if (scheduler.schedule_callback == nullptr) {
     scheduler = default_scheduler_;
@@ -341,8 +369,7 @@ void Engine::SetDefaultMessageScheduler(DartEngine_MessageScheduler scheduler) {
 
 void Engine::SetMessageScheduler(DartEngine_MessageScheduler scheduler,
                                  Dart_Isolate isolate) {
-  MutexLocker ml(&engine_state_);
-  schedulers_[isolate] = scheduler;
+  DataForIsolate(isolate)->scheduler = scheduler;
 }
 
 }  // namespace engine

runtime/engine/engine.h

@@ -6,7 +6,6 @@
 #define RUNTIME_ENGINE_ENGINE_H_
 
 #include <memory>
-#include <queue>
 #include <unordered_map>
 #include <vector>
 #include "include/dart_engine.h"
@@ -54,6 +53,9 @@ class Engine {
   // Calls Dart_HandleMessage, managing an isolate lock and Dart scope.
   void HandleMessage(Dart_Isolate isolate);
 
+  // Drains the microtasks queue, requires an active isolate.
+  Dart_Handle DrainMicrotasksQueue();
+
   // Sets a callback to be called when Dart_HandleMessage returns an error.
   void SetHandleMessageErrorCallback(
       DartEngine_HandleMessageErrorCallback callback);
@@ -85,6 +87,14 @@ class Engine {
   static void HandleMessageCallback(Dart_Isolate isolate);
 
  private:
+  // Engine's internal data for isolate.
+  struct IsolateData {
+    DartEngine_MessageScheduler scheduler;
+    Mutex mutex;
+    Dart_PersistentHandle isolate_library;
+    Dart_PersistentHandle drain_microtasks_function_name;
+  };
+
   // Set to false once shutdown starts.
   bool is_running_ = false;
 
@@ -114,19 +124,17 @@ class Engine {
   // All isolates, started via Engine::StartIsolate.
   std::vector<Dart_Isolate> isolates_;
 
-  // Stores per-isolate mutexes, used by Engine::LockIsolate/UnlockIsolate.
-  std::unordered_map<Dart_Isolate, Mutex> mutexes_;
+  // Stores per-isolate engine state.
+  std::unordered_map<Dart_Isolate, std::shared_ptr<IsolateData>> isolate_data_;
 
   // Default scheduler.
   DartEngine_MessageScheduler default_scheduler_;
-  // Per-isolate message schedulers.
-  std::unordered_map<Dart_Isolate, DartEngine_MessageScheduler> schedulers_;
 
   // Callback to notify about Dart_HandleMessage errors.
   DartEngine_HandleMessageErrorCallback handle_message_error_callback_;
 
-  // Helper function to get an element from mutexes_.
-  Mutex& MutexForIsolate(Dart_Isolate isolate);
+  // Helper function to get an element from isolate_data_.
+  std::shared_ptr<IsolateData> DataForIsolate(Dart_Isolate isolate);
 };
 
 }  // namespace engine

runtime/engine/include/dart_engine.h

@@ -109,6 +109,17 @@ typedef void (*DartEngine_HandleMessageErrorCallback)(
 DART_EXPORT void DartEngine_SetHandleMessageErrorCallback(
     DartEngine_HandleMessageErrorCallback handle_message_error_callback);
 
+/**
+ * Drains the microtasks queue. Requires to be an active isolate.
+ *
+ * Normally the microtasks queue is drained after handling each
+ * isolate message, but when the engine calls into Dart, it might be
+ * required to manually drain the microtasks queue.
+ *
+ * \return Dart_Handle invocation result.
+ */
+DART_EXPORT Dart_Handle DartEngine_DrainMicrotasksQueue();
+
 /**
  * Handles a single message for an isolate.
  */

runtime/tools/entitlements/run_futures_aot.plist

@@ -0,0 +1,10 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple Computer//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+    <dict>
+        <key>com.apple.security.cs.allow-jit</key>
+        <true/>
+        <key>com.apple.security.cs.disable-library-validation</key>
+        <true/>
+    </dict>
+</plist>

runtime/tools/entitlements/run_futures_kernel.plist

@@ -0,0 +1,10 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple Computer//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+    <dict>
+        <key>com.apple.security.cs.allow-jit</key>
+        <true/>
+        <key>com.apple.security.cs.disable-library-validation</key>
+        <true/>
+    </dict>
+</plist>

samples/embedder/BUILD.gn

@@ -15,6 +15,7 @@ group("all") {
 
 group("aot") {
   deps = [
+    ":run_futures_aot",
     ":run_main_aot",
     ":run_timer_aot",
     ":run_timer_async_aot",
@@ -24,6 +25,7 @@ group("aot") {
 
 group("kernel") {
   deps = [
+    ":run_futures_kernel",
     ":run_main_kernel",
     ":run_timer_async_kernel",
     ":run_timer_kernel",
@@ -150,3 +152,12 @@ sample("run_timer_async") {
 snapshots("timer") {
   main_dart = "timer.dart"
 }
+
+snapshots("futures") {
+  main_dart = "futures.dart"
+}
+
+sample("run_futures") {
+  sources = [ "run_futures.cc" ]
+  snapshots = [ ":futures" ]
+}

samples/embedder/futures.dart

@@ -0,0 +1,129 @@
+// Copyright (c) 2025, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:ffi';
+
+void main() async {
+  print(
+    'returnRegularFuture returns: '
+    '${await returnRegularFuture(5)}',
+  );
+  print(
+    'sumIntStream(useAsyncStar = false) returns: '
+    '${await sumIntStream(5, 1, false)}',
+  );
+  print(
+    'sumIntStream(useAsyncStar = true) returns: '
+    '${await sumIntStream(5, 1, true)}',
+  );
+}
+
+Future<int> returnRegularFuture(int delayMs) async {
+  await Future.delayed(Duration(milliseconds: 5));
+  return 256;
+}
+
+Future<int> returnMicrotaskFuture(int delayMs) => Future.microtask(() async {
+  await Future.delayed(Duration(milliseconds: delayMs));
+  return 256;
+});
+
+Stream<int> produceIntStreamWithAsyncStar(int count, int delayMs) async* {
+  final delay = Duration(milliseconds: delayMs);
+
+  for (var i = 0; i < count; i++) {
+    await Future.delayed(delay);
+    yield i;
+  }
+}
+
+Stream<int> produceIntStreamWithController(int count, int delayMs) {
+  final delay = Duration(milliseconds: delayMs);
+
+  final sc = StreamController<int>();
+  (() async {
+    for (var i = 0; i < count; i++) {
+      await Future.delayed(delay);
+      sc.add(i);
+    }
+    sc.close();
+  })();
+  return sc.stream;
+}
+
+Future<int> sumIntStream(int count, int delayMs, bool useAsyncStar) async {
+  final stream =
+      useAsyncStar
+          ? produceIntStreamWithAsyncStar(count, delayMs)
+          : produceIntStreamWithController(count, delayMs);
+  var sum = 0;
+  await for (var value in stream) {
+    sum += value;
+  }
+  return sum;
+}
+
+Future<int> awaitAndMultiply(Future<int> a, Future<int> b) async =>
+    (await a) * (await b);
+
+class AwaitAndMultiplyCall {
+  final _a = Completer<int>();
+  final _b = Completer<int>();
+
+  Future<int> getA() => _a.future;
+  Future<int> getB() => _b.future;
+
+  @pragma('vm:entry-point', 'call')
+  void setA(int value) => _a.complete(value);
+  @pragma('vm:entry-point', 'call')
+  void setB(int value) => _b.complete(value);
+}
+
+@pragma('vm:entry-point', 'call')
+AwaitAndMultiplyCall awaitAndMultiplyC(int callbackPtr, int contextPtr) {
+  final call = AwaitAndMultiplyCall();
+  () async {
+    Pointer<NativeFunction<Void Function(Pointer<Opaque>, Int64)>>.fromAddress(
+      callbackPtr,
+    ).asFunction<void Function(Pointer<Opaque>, int)>()(
+      Pointer<Opaque>.fromAddress(contextPtr),
+      await awaitAndMultiply(call.getA(), call.getB()),
+    );
+  }();
+  return call;
+}
+
+@pragma('vm:entry-point', 'call')
+// C-friendly wrapper over [returnRegularFuture].
+void returnRegularFutureC(
+  int delayMs,
+  bool useMicrotask,
+  int callbackPtr,
+  int contextPtr,
+) async =>
+    Pointer<NativeFunction<Void Function(Pointer<Opaque>, Int64)>>.fromAddress(
+      callbackPtr,
+    ).asFunction<void Function(Pointer<Opaque>, int)>()(
+      Pointer<Opaque>.fromAddress(contextPtr),
+      await (useMicrotask
+          ? returnMicrotaskFuture(delayMs)
+          : returnRegularFuture(delayMs)),
+    );
+
+@pragma('vm:entry-point', 'call')
+// C-friendly wrapper over [sumIntStream].
+void sumIntStreamC(
+  int count,
+  int delayMs,
+  bool useAsyncStar,
+  int callbackPtr,
+  int contextPtr,
+) async =>
+    Pointer<NativeFunction<Void Function(Pointer<Opaque>, Int64)>>.fromAddress(
+      callbackPtr,
+    ).asFunction<void Function(Pointer<Opaque>, int)>()(
+      Pointer<Opaque>.fromAddress(contextPtr),
+      await sumIntStream(count, delayMs, useAsyncStar),
+    );