[vm/shared] Implement Isolate runEventLoopSync.

TEST=ffi/threading_runeventloop_test

Change-Id: I2061d10d14bcdd589a4a9893f122620f8d0a4e7e
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/497100
Reviewed-by: Slava Egorov <vegorov@google.com>

runtime/lib/isolate.cc

@@ -569,6 +569,18 @@ class IsolateAcquireScope : public ValueObject {
   void Reset() { isolate_ = nullptr; }
   Isolate* isolate() { return isolate_; }
   IsolateAcquireResult acquire_result() { return acquire_result_; }
+  const char* error_message() {
+    switch (acquire_result_) {
+      case IsolateAcquireResult::ISOLATE_NOT_AVAILABLE:
+        return "Unable to enter the isolate as it's unavailable";
+      case IsolateAcquireResult::PINNED_TO_ANOTHER_THREAD:
+        return "Isolate is pinned to a different thread already";
+      case IsolateAcquireResult::BUSY:
+        return "Isolate is busy, running on a different thread";
+      default:
+        UNREACHABLE();
+    }
+  }
 
  private:
   Isolate* isolate_;
@@ -605,6 +617,72 @@ DEFINE_NATIVE_ENTRY(Isolate_shutdownSync_, 0, 1) {
   return Object::null();
 }
 
+DEFINE_NATIVE_ENTRY(Isolate_runEventLoopSync_, 0, 1) {
+  GET_NON_NULL_NATIVE_ARGUMENT(SendPort, isolate_control_port,
+                               arguments->NativeArgAt(0));
+  if (thread->isolate() != nullptr) {
+    const auto& error =
+        String::Handle(String::New("Should be invoked outside of an isolate"));
+    Exceptions::ThrowStateError(error);
+    UNREACHABLE();
+  }
+
+  auto group = thread->isolate_group();
+
+  if (isolate_control_port.origin_id() != group->id()) {
+    const auto& error = String::Handle(String::New(
+        "Target isolate should be part of the same isolate group."));
+    Exceptions::ThrowStateError(error);
+    UNREACHABLE();
+  }
+
+  Dart_Port control_port_id = isolate_control_port.Id();
+
+  if (PortMap::HasEventLoopRunning(control_port_id)) {
+    const auto& error =
+        String::Handle(String::New("Isolate has a message loop running."));
+    Exceptions::ThrowStateError(error);
+    UNREACHABLE();
+  }
+
+  Error& result_error = Error::Handle();
+  Thread::ExitIsolateGroupAsMutator(/*bypass_safepoint=*/false);
+  {
+    // Take over isolate's event loop - block it's original message_handler.
+    IsolateAcquireScope acquire_scope(thread, control_port_id);
+    Isolate* target_isolate = acquire_scope.isolate();
+    if (target_isolate == nullptr) {
+      // Reenter the group so we can report an error.
+      Thread::EnterIsolateGroupAsMutator(group, /*bypass_safepoint=*/false,
+                                         thread);
+      const auto& message =
+          String::Handle(String::New(acquire_scope.error_message()));
+      Exceptions::ThrowStateError(message);
+      UNREACHABLE();
+    }
+    auto current_thread = Thread::Current();
+
+    {
+      TransitionVMToNative transition(current_thread);
+
+      Dart_EnterScope();
+      Dart_ExitIsolate();
+      target_isolate->message_handler()->RunSync();
+      Dart_EnterIsolate(reinterpret_cast<Dart_Isolate>(target_isolate));
+      if (target_isolate->sticky_error() != Object::null()) {
+        result_error = target_isolate->StealStickyError();
+      }
+      Dart_ExitScope();
+    }
+  }
+  Thread::EnterIsolateGroupAsMutator(group, /*bypass_safepoint=*/false, thread);
+  if (!result_error.IsNull()) {
+    Exceptions::PropagateError(result_error);
+    UNREACHABLE();
+  }
+  return Object::null();
+}
+
 DEFINE_NATIVE_ENTRY(Isolate_runSync_, 1, 2) {
   GET_NON_NULL_NATIVE_ARGUMENT(SendPort, isolate_control_port,
                                arguments->NativeArgAt(0));
@@ -661,20 +739,7 @@ DEFINE_NATIVE_ENTRY(Isolate_runSync_, 1, 2) {
           Thread::EnterIsolateGroupAsMutator(group, /*bypass_safepoint=*/false,
                                              thread);
         }
-        const char* message;
-        switch (acquire_scope.acquire_result()) {
-          case IsolateAcquireResult::ISOLATE_NOT_AVAILABLE:
-            message = "Unable to enter the isolate as it's unavailable";
-            break;
-          case IsolateAcquireResult::PINNED_TO_ANOTHER_THREAD:
-            message = "Isolate is pinned to a different thread already";
-            break;
-          case IsolateAcquireResult::BUSY:
-            message = "Isolate is busy, running on a different thread";
-            break;
-          default:
-            UNREACHABLE();
-        }
+        const char* message = acquire_scope.error_message();
         Exceptions::ThrowStateError(String::Handle(String::New(message)));
         UNREACHABLE();
       }

runtime/vm/bootstrap_natives.h

@@ -259,6 +259,7 @@ namespace dart {
   V(Isolate_getDebugName, 1)                                                   \
   V(Isolate_getPortAndCapabilitiesOfCurrentIsolate, 0)                         \
   V(Isolate_runSync_, 2)                                                       \
+  V(Isolate_runEventLoopSync_, 1)                                              \
   V(Isolate_sendOOB, 2)                                                        \
   V(Isolate_shutdownSync_, 1)                                                  \
   V(Isolate_spawnFunction, 10)                                                 \

runtime/vm/message_handler.cc

@@ -113,6 +113,12 @@ bool MessageHandler::Run(ThreadPool* pool,
   return result;
 }
 
+void MessageHandler::RunSync() {
+  task_running_ = true;
+  TaskCallback();
+  task_running_ = false;
+}
+
 void MessageHandler::PostMessage(std::unique_ptr<Message> message,
                                  bool before_events) {
   Message::Priority saved_priority;

runtime/vm/message_handler.h

@@ -43,10 +43,19 @@ class MessageHandler : public PortHandler {
   // HandleMessage() indicates that an error has occurred during
   // message processing.
 
-  // Returns false if the handler terminated abnormally, otherwise it
-  // returns true.
+  // Returns false if the handler failed to launch on the thread pool(due to
+  // thread pool shutting down for example), otherwise it returns true.
   bool Run(ThreadPool* pool, EndCallback end_callback, CallbackData data);
 
+  // Runs this message handler on current thread.
+  //
+  // A message handler will run until it terminates either normally or
+  // abnormally.  Normal termination occurs when the message handler
+  // no longer has any live ports.  Abnormal termination occurs when
+  // HandleMessage() indicates that an error has occurred during
+  // message processing.
+  void RunSync();
+
   // Handles the next message for this message handler.  Should only
   // be used when not running the handler on the thread pool (via Run
   // or RunBlocking).

sdk/lib/_internal/vm/lib/isolate_patch.dart

@@ -765,7 +765,7 @@ final class Isolate {
 
   @patch
   void runEventLoopSync() {
-    throw UnsupportedError("Isolate.runEventLoopSync");
+    _runEventLoopSync(controlPort);
   }
 
   @patch
@@ -778,6 +778,9 @@ final class Isolate {
     throw UnsupportedError("Isolate.isPinnedToCurrentThread");
   }
 
+  @pragma("vm:external-name", "Isolate_runEventLoopSync_")
+  external static void _runEventLoopSync(SendPort controlPort);
+
   @patch
   void set onEvent(void Function(Isolate) callback) {
     throw UnsupportedError("Isolate.onEvent");

tests/ffi/threading_runeventloop_test.dart

@@ -0,0 +1,165 @@
+// Copyright (c) 2026, 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.
+//
+// Tests Isolate threading API.
+//
+// VMOptions=--experimental-shared-data
+
+import 'dart:async';
+import 'dart:concurrent';
+import 'dart:ffi';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:dart_internal/isolate_group.dart' show IsolateGroup;
+import "package:expect/async_helper.dart";
+import 'package:expect/expect.dart';
+import 'package:ffi/ffi.dart';
+
+import 'threading_utils.dart';
+
+@pragma('vm:shared')
+int counter = 0;
+
+int foo = 42;
+
+@pragma('vm:shared')
+late Mutex mutexCondvar;
+@pragma('vm:shared')
+late ConditionVariable condVar;
+@pragma('vm:shared')
+int greetingsReceived = 0;
+
+int threadMain(Pointer<Void> data) {
+  final pthreadSelf = DynamicLibrary.process()
+      .lookupFunction<PthreadSelfNFT, PthreadSelfFT>('pthread_self');
+  final self = pthreadSelf();
+  final i = data.cast<Uint8>()[0];
+  print('threadMain started with $data i:$i pthreadid $self');
+  final new_isolate = Isolate.create(debugName: "helper");
+  Expect.isNotNull(new_isolate);
+  final SendPort sp = new_isolate.runSync(() {
+    late RawReceivePort rp;
+    rp = RawReceivePort((e) {
+      print('running RawReceivePort handler $e');
+      final pthreadSelf = DynamicLibrary.process()
+          .lookupFunction<PthreadSelfNFT, PthreadSelfFT>('pthread_self');
+      final isolate_self = pthreadSelf();
+      print('=== receivePort handler received $e on pthreadid $self');
+      Expect.equals(self, isolate_self);
+
+      Expect.equals("greetings!", e);
+      mutexCondvar.runLocked(() {
+        greetingsReceived |= (1 << i);
+        condVar.notify();
+      });
+      rp.close();
+    });
+    return rp.sendPort;
+  });
+
+  Expect.isNotNull(sp);
+  sp.send('greetings!');
+
+  // No response is expected until we start running event loop.
+  mutexCondvar.runLocked(() => condVar.wait(mutexCondvar, /*timeout_ms=*/ 100));
+  Expect.isFalse(((1 << i) & greetingsReceived) != 0);
+
+  print('=== running event loop for $new_isolate');
+  new_isolate.runEventLoopSync();
+  mutexCondvar.runLocked(() {
+    while (((1 << i) & greetingsReceived) == 0) {
+      condVar.wait(mutexCondvar);
+    }
+  });
+  Expect.isTrue(((1 << i) & greetingsReceived) != 0);
+
+  print('=== running runSync again');
+  new_isolate.runSync(() {
+    print('=== hi, kuka ${++foo}!');
+    Expect.equals(43, foo);
+  });
+
+  print('=== shutting down');
+  new_isolate.shutdownSync();
+  return 0;
+}
+
+ThreadInfo testRunOnNewIsolateOnNewThread(
+  int i,
+  int Function(Pointer<Void>) threadMain,
+) {
+  final threadInfo = ThreadInfo();
+
+  Expect.equals(0, pthreadAttrInit(threadInfo.ptr_attr));
+  threadInfo.ptr_data.cast<Uint8>()[0] = i;
+  print(
+    '=== ptr_data: ${threadInfo.ptr_data.address.toRadixString(16)}, i: $i',
+  );
+  final callback =
+      NativeCallable<IntPtr Function(Pointer<Void>)>.isolateGroupBound(
+        threadMain,
+        exceptionalReturn: -1,
+      );
+  callback.keepIsolateAlive = false;
+  pthreadCreate(
+    threadInfo.ptr_tid,
+    threadInfo.ptr_attr,
+    callback.nativeFunction,
+    threadInfo.ptr_data.cast<Void>(),
+  );
+  return threadInfo;
+}
+
+Future<void> testRunEventLoopManyThreads({int numThreads = 63}) async {
+  if (Platform.isWindows) {
+    // pthread library loading doesn't work on Windows.
+    return;
+  }
+  mutexCondvar = Mutex();
+  condVar = ConditionVariable();
+  final threadInfos = <ThreadInfo>[];
+  final repliedMask = (1 << numThreads) - 1;
+  print('repliedMask: ${repliedMask.toRadixString(16)}');
+  for (int i = 0; i < numThreads; i++) {
+    threadInfos.add(testRunOnNewIsolateOnNewThread(i, threadMain));
+  }
+  mutexCondvar.runLocked(() {
+    while (greetingsReceived < repliedMask) {
+      condVar.wait(mutexCondvar);
+      print('main received ${greetingsReceived.toRadixString(16)}');
+    }
+  });
+  print('main is happy received ${greetingsReceived.toRadixString(16)}');
+
+  for (ThreadInfo threadInfo in threadInfos) {
+    threadInfo.join();
+  }
+}
+
+Future<void> testFailRunEventLoopFromIsolate() async {
+  Expect.throws(
+    () {
+      Isolate.current.runEventLoopSync();
+    },
+    (e) =>
+        e is StateError &&
+        e.message.contains("Should be invoked outside of an isolate"),
+  );
+}
+
+main(List<String> args, SendPort? message) async {
+  if (message != null) {
+    Expect.equals(1, args.length);
+    Expect.equals("worker", args[0]);
+    await ReceivePort().first;
+    return;
+  }
+  asyncStart();
+
+  await testRunEventLoopManyThreads(numThreads: 63);
+  await testFailRunEventLoopFromIsolate();
+
+  asyncEnd();
+}

tests/ffi/threading_test.dart

@@ -18,30 +18,7 @@ import 'package:expect/expect.dart';
 import 'package:ffi/ffi.dart';
 
 import 'dylib_utils.dart';
-
-typedef PthreadAttrInitFT = int Function(Pointer<Char>);
-typedef PthreadAttrInitNFT = IntPtr Function(Pointer<Char>);
-final pthreadAttrInit = DynamicLibrary.process()
-    .lookupFunction<PthreadAttrInitNFT, PthreadAttrInitFT>('pthread_attr_init');
-
-typedef PthreadAttrDestroyFT = int Function(Pointer<Char>);
-typedef PthreadAttrDestroyNFT = IntPtr Function(Pointer<Char>);
-final pthreadAttrDestroy = DynamicLibrary.process()
-    .lookupFunction<PthreadAttrDestroyNFT, PthreadAttrDestroyFT>(
-      'pthread_attr_destroy',
-    );
-
-typedef PthreadCreateFT =
-    int Function(Pointer<IntPtr>, Pointer<Char>, Pointer, Pointer<Void>);
-typedef PthreadCreateNFT =
-    IntPtr Function(Pointer<IntPtr>, Pointer<Char>, Pointer, Pointer<Void>);
-final pthreadCreate = DynamicLibrary.process()
-    .lookupFunction<PthreadCreateNFT, PthreadCreateFT>('pthread_create');
-
-typedef PthreadJoinFT = int Function(int, Pointer<Void>);
-typedef PthreadJoinNFT = IntPtr Function(IntPtr, Pointer<Void>);
-final pthreadJoin = DynamicLibrary.process()
-    .lookupFunction<PthreadJoinNFT, PthreadJoinFT>('pthread_join');
+import 'threading_utils.dart';
 
 @pragma('vm:shared')
 int counter = 0;
@@ -163,24 +140,6 @@ int threadMain(Pointer<Void> data) {
   return 0;
 }
 
-class ThreadInfo {
-  final ptr_attr = calloc<Char>(64); // big enough to fit pthread_attr_t?
-  final ptr_tid = calloc<IntPtr>(1);
-  final ptr_data = calloc<Int32>(1024);
-  final ptr_retval = calloc<IntPtr>(1024);
-
-  void join() {
-    Expect.equals(0, pthreadJoin(ptr_tid.value, ptr_retval.cast<Void>()));
-    calloc.free(ptr_retval);
-
-    calloc.free(ptr_data);
-    calloc.free(ptr_tid);
-
-    Expect.equals(0, pthreadAttrDestroy(ptr_attr));
-    calloc.free(ptr_attr);
-  }
-}
-
 Future<void> testRunSyncOnPinnedToSelfIsolate() async {
   if (Platform.isWindows) {
     return; // pthread is not available on Windows.

tests/ffi/threading_utils.dart

@@ -0,0 +1,56 @@
+// Copyright (c) 2026, 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:ffi';
+
+import 'package:expect/expect.dart';
+import 'package:ffi/ffi.dart';
+
+typedef PthreadAttrInitFT = int Function(Pointer<Char>);
+typedef PthreadAttrInitNFT = IntPtr Function(Pointer<Char>);
+final pthreadAttrInit = DynamicLibrary.process()
+    .lookupFunction<PthreadAttrInitNFT, PthreadAttrInitFT>('pthread_attr_init');
+
+typedef PthreadAttrDestroyFT = int Function(Pointer<Char>);
+typedef PthreadAttrDestroyNFT = IntPtr Function(Pointer<Char>);
+final pthreadAttrDestroy = DynamicLibrary.process()
+    .lookupFunction<PthreadAttrDestroyNFT, PthreadAttrDestroyFT>(
+      'pthread_attr_destroy',
+    );
+
+typedef PthreadCreateFT =
+    int Function(Pointer<IntPtr>, Pointer<Char>, Pointer, Pointer<Void>);
+typedef PthreadCreateNFT =
+    IntPtr Function(Pointer<IntPtr>, Pointer<Char>, Pointer, Pointer<Void>);
+final pthreadCreate = DynamicLibrary.process()
+    .lookupFunction<PthreadCreateNFT, PthreadCreateFT>('pthread_create');
+
+typedef PthreadJoinFT = int Function(int, Pointer<Void>);
+typedef PthreadJoinNFT = IntPtr Function(IntPtr, Pointer<Void>);
+final pthreadJoin = DynamicLibrary.process()
+    .lookupFunction<PthreadJoinNFT, PthreadJoinFT>('pthread_join');
+
+typedef PthreadSelfFT = int Function();
+typedef PthreadSelfNFT = IntPtr Function();
+final pthreadSelf = DynamicLibrary.process()
+    .lookupFunction<PthreadSelfNFT, PthreadSelfFT>('pthread_self');
+
+class ThreadInfo {
+  final ptr_attr = calloc<Char>(64); // big enough to fit pthread_attr_t?
+  final ptr_tid = calloc<IntPtr>(1);
+  final ptr_data = calloc<Int32>(1024);
+  final ptr_retval = calloc<IntPtr>(1024);
+
+  void join() {
+    Expect.equals(0, pthreadJoin(ptr_tid.value, ptr_retval.cast<Void>()));
+    calloc.free(ptr_retval);
+
+    calloc.free(ptr_data);
+    calloc.free(ptr_tid);
+
+    Expect.equals(0, pthreadAttrDestroy(ptr_attr));
+    calloc.free(ptr_attr);
+  }
+}