[vm/shared] Introduce isolate event loop handling dart api.

Implement api methods to create and shutdown isolates from IsolateGroup-bound callbacks (normally invoked from native code), run dart code on such isolates.
The rest of the api is not implemented yet.

TEST=tests/ffi/threading_test.dart
BUG=https://github.com/dart-lang/sdk/issues/62407
CoreLibraryReviewExempt: vm-only change to isolate library
Change-Id: I0271ead8ba011dfe9d7953769415d6a88a962854
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/486522
Commit-Queue: Alexander Aprelev <aam@google.com>
Reviewed-by: Ryan Macnak <rmacnak@google.com>

runtime/lib/isolate.cc

@@ -496,6 +496,220 @@ class MessageValidator : private WorkSet {
   ClassTable* class_table_;
 };
 
+DEFINE_NATIVE_ENTRY(Isolate_create_, 0, 1) {
+  GET_NON_NULL_NATIVE_ARGUMENT(String, debug_name, arguments->NativeArgAt(0));
+  const char* debug_name_cstr =
+      !debug_name.IsNull() ? debug_name.ToCString() : nullptr;
+
+  if (thread->isolate() != nullptr) {
+    const auto& error =
+        String::Handle(String::New("Should be invoked outside of an isolate"));
+    Exceptions::ThrowStateError(error);
+    UNREACHABLE();
+  }
+
+  char* error = nullptr;
+  auto group = IsolateGroup::Current();
+
+  auto& created = Array::Handle(zone, Array::New(2));
+  const auto& capabilities = Array::Handle(zone, Array::New(2));
+  auto& capability = Capability::Handle(zone);
+  auto& send_port = SendPort::Handle(zone);
+
+  Thread::ExitIsolateGroupAsMutator(/*bypass_safepoint=*/false);
+  Isolate* created_isolate =
+      CreateWithinExistingIsolateGroup(group, debug_name_cstr, &error);
+  RELEASE_ASSERT(created_isolate != nullptr);
+
+  Dart_ExitIsolate();
+  Thread::EnterIsolateGroupAsMutator(group, /*bypass_safepoint=*/false, thread);
+
+  auto current_thread = Thread::Current();
+  StackZone stack_zone(current_thread);
+
+  capability = Capability::New(created_isolate->pause_capability());
+  capabilities.SetAt(0, capability);
+  capability = Capability::New(created_isolate->terminate_capability());
+  capabilities.SetAt(1, capability);
+  send_port = SendPort::New(created_isolate->main_port(), group->id());
+  created.SetAt(0, send_port);
+  created.SetAt(1, capabilities);
+  return created.ptr();
+}
+
+namespace {
+
+// Scope that gives user an ability to acquire an isolate if it is available,
+// gain temporary ownership, which is released at the end of the scope.
+class IsolateAcquireScope : public ValueObject {
+ public:
+  IsolateAcquireScope(Thread* thread, Dart_Port receiver) : isolate_(nullptr) {
+    Isolate* target_isolate = nullptr;
+    acquire_result_ =
+        PortMap::AcquireIsolateByControlPort(receiver, &target_isolate);
+    if (target_isolate == nullptr) {
+      // Isolate might have exited already.
+      return;
+    }
+    Thread::EnterIsolate(target_isolate);
+    isolate_ = target_isolate;
+  }
+
+  ~IsolateAcquireScope() {
+    if (isolate_ == nullptr) {
+      return;
+    }
+    ASSERT(Thread::Current()->isolate() == isolate_);
+    Thread::ExitIsolate();
+    if (!isolate_->is_permanently_pinned()) {
+      isolate_->ReleaseOwnership();
+    }
+  }
+
+  void Reset() { isolate_ = nullptr; }
+  Isolate* isolate() { return isolate_; }
+  IsolateAcquireResult acquire_result() { return acquire_result_; }
+
+ private:
+  Isolate* isolate_;
+  IsolateAcquireResult acquire_result_;
+
+  DISALLOW_COPY_AND_ASSIGN(IsolateAcquireScope);
+};
+
+}  // namespace
+
+DEFINE_NATIVE_ENTRY(Isolate_shutdownSync_, 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();
+
+  auto control_port_id = isolate_control_port.Id();
+
+  Thread::ExitIsolateGroupAsMutator(/*bypass_safepoint=*/false);
+  {
+    IsolateAcquireScope acquire_scope(thread, control_port_id);
+    Isolate* target_isolate = acquire_scope.isolate();
+    if (target_isolate != nullptr) {
+      Dart::ShutdownIsolate(Thread::Current());
+      acquire_scope.Reset();
+    }
+  }
+  Thread::EnterIsolateGroupAsMutator(group, /*bypass_safepoint=*/false, thread);
+  return Object::null();
+}
+
+DEFINE_NATIVE_ENTRY(Isolate_runSync_, 1, 2) {
+  GET_NON_NULL_NATIVE_ARGUMENT(SendPort, isolate_control_port,
+                               arguments->NativeArgAt(0));
+  GET_NON_NULL_NATIVE_ARGUMENT(Closure, closure, arguments->NativeArgAt(1));
+  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();
+
+  closure.EnsureDeeplyImmutable(zone);
+
+  Array& args_desc = Array::Handle(zone, ArgumentsDescriptor::NewBoxed(0, 1));
+  Array& args = Array::Handle(zone, Array::New(1));
+  args.SetAt(0, closure);
+  Object& result = Object::Handle(zone);
+
+  if (isolate != nullptr && (isolate->main_port() == control_port_id)) {
+    // Fast-path for when we are already running on the desired isolate.
+    result = DartEntry::InvokeClosure(thread, args, args_desc);
+
+    if (result.IsUnwindError()) {
+      Exceptions::PropagateError(Error::Cast(result));
+      UNREACHABLE();
+    }
+  } else {
+    if (PortMap::HasEventLoopRunning(control_port_id)) {
+      const auto& error =
+          String::Handle(String::New("Isolate has a message loop running."));
+      Exceptions::ThrowStateError(error);
+      UNREACHABLE();
+    }
+
+    if (isolate != nullptr) {
+      ASSERT(Thread::Current()->isolate() == isolate);
+      Thread::ExitIsolate(/*isolate_shutdown=*/false);
+    } else {
+      Thread::ExitIsolateGroupAsMutator(/*bypass_safepoint=*/false);
+    }
+
+    {
+      IsolateAcquireScope acquire_scope(thread, control_port_id);
+      Isolate* target_isolate = acquire_scope.isolate();
+      if (target_isolate == nullptr) {
+        // Re-enter the group or isolate so we can report an error.
+        if (isolate != nullptr) {
+          Thread::EnterIsolate(isolate);
+        } else {
+          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();
+        }
+        Exceptions::ThrowStateError(String::Handle(String::New(message)));
+        UNREACHABLE();
+      }
+
+      auto current_thread = Thread::Current();
+      {
+        StackZone stack_zone(current_thread);
+        result = DartEntry::InvokeClosure(current_thread, args, args_desc);
+      }
+    }
+
+    if (isolate != nullptr) {
+      Thread::EnterIsolate(isolate);
+    } else {
+      Thread::EnterIsolateGroupAsMutator(group, /*bypass_safepoint=*/false,
+                                         thread);
+    }
+
+    if (result.IsUnwindError()) {
+      const auto& error =
+          String::Handle(String::New("Isolate was forced to exit."));
+      Exceptions::ThrowStateError(error);
+      UNREACHABLE();
+    }
+  }
+  if (result.IsUnhandledException()) {
+    Exceptions::PropagateError(Error::Cast(result));
+    UNREACHABLE();
+  }
+
+  result.EnsureDeeplyImmutable(zone);
+
+  return result.ptr();
+}
+
 // TODO(http://dartbug.com/47777): Add support for Finalizers.
 DEFINE_NATIVE_ENTRY(Isolate_exit_, 0, 2) {
   if (isolate == nullptr) {
@@ -992,8 +1206,8 @@ class SpawnIsolateTask : public ThreadPool::Task {
     capabilities.SetAt(0, capability);
     capability = Capability::New(isolate->terminate_capability());
     capabilities.SetAt(1, capability);
-    const auto& send_port =
-        SendPort::Handle(zone, SendPort::New(isolate->main_port()));
+    const auto& send_port = SendPort::Handle(
+        zone, SendPort::New(isolate->main_port(), isolate->group()->id()));
     const auto& message = Array::Handle(zone, Array::New(2));
     message.SetAt(0, send_port);
     message.SetAt(1, capabilities);
@@ -1234,7 +1448,8 @@ DEFINE_NATIVE_ENTRY(Isolate_getDebugName, 0, 1) {
 
 DEFINE_NATIVE_ENTRY(Isolate_getPortAndCapabilitiesOfCurrentIsolate, 0, 0) {
   const Array& result = Array::Handle(Array::New(3));
-  result.SetAt(0, SendPort::Handle(SendPort::New(isolate->main_port())));
+  result.SetAt(0, SendPort::Handle(SendPort::New(isolate->main_port(),
+                                                 isolate->group()->id())));
   result.SetAt(
       1, Capability::Handle(Capability::New(isolate->pause_capability())));
   result.SetAt(

runtime/vm/bootstrap_natives.h

@@ -253,11 +253,14 @@ namespace dart {
   V(Int32x4_setFlagZ, 2)                                                       \
   V(Int32x4_setFlagW, 2)                                                       \
   V(Int32x4_select, 3)                                                         \
+  V(Isolate_create_, 1)                                                        \
   V(Isolate_exit_, 2)                                                          \
   V(Isolate_getCurrentRootUriStr, 0)                                           \
   V(Isolate_getDebugName, 1)                                                   \
   V(Isolate_getPortAndCapabilitiesOfCurrentIsolate, 0)                         \
+  V(Isolate_runSync_, 2)                                                       \
   V(Isolate_sendOOB, 2)                                                        \
+  V(Isolate_shutdownSync_, 1)                                                  \
   V(Isolate_spawnFunction, 10)                                                 \
   V(Isolate_spawnUri, 12)                                                      \
   V(GrowableList_allocate, 1)                                                  \

runtime/vm/dart_api_impl.cc

@@ -2166,9 +2166,23 @@ DART_EXPORT void Dart_SetCurrentThreadOwnsIsolate() {
   CHECK_ISOLATE(isolate);
   if (!isolate->SetOwnerThread(OSThread::kInvalidThreadId,
                                OSThread::GetCurrentThreadId())) {
-    FATAL("Tried to claim ownership of isolate %s, but it is already owned\n",
+    // We might be running this method while running dart code
+    // on this target isolate.
+    // So first confirm that the isolate is not pinned yet.
+    if (isolate->is_permanently_pinned()) {
+      FATAL("Tried to claim ownership of isolate %s, but it is already owned\n",
+            isolate->name());
+    }
+    // Allow pinning only if current owner is the current thread.
+    if (isolate->GetOwnerThread(/*locker=*/nullptr) !=
+        OSThread::GetCurrentThreadId()) {
+      FATAL(
+          "Tried to claim ownership of isolate %s, but it is running on"
+          "some other thread\n",
           isolate->name());
+    }
   }
+  isolate->set_is_permanently_pinned();
 }
 
 DART_EXPORT void Dart_ClearCurrentThreadOwnsIsolate_ForTesting() {
@@ -2179,6 +2193,7 @@ DART_EXPORT void Dart_ClearCurrentThreadOwnsIsolate_ForTesting() {
     FATAL("Tried to clear ownership of isolate %s, but we don't own it\n",
           isolate->name());
   }
+  isolate->clear_is_permanently_pinned_for_testing_only();
 }
 
 DART_EXPORT bool Dart_GetCurrentThreadOwnsIsolate(Dart_Port port) {

runtime/vm/isolate.cc

@@ -1164,8 +1164,6 @@ class IsolateMessageHandler : public MessageHandler {
   }
 
  private:
-  // A result of false indicates that the isolate should terminate the
-  // processing of further events.
   ErrorPtr HandleLibMessage(const Array& message);
 
   MessageStatus ProcessUnhandledException(const Error& result);
@@ -3778,6 +3776,20 @@ void Isolate::WaitForOutstandingSpawns() {
   }
 }
 
+bool Isolate::TryAcquireOwnership() {
+  ThreadId current_thread_id = OSThread::GetCurrentThreadId();
+  if (SetOwnerThread(OSThread::kInvalidThreadId, current_thread_id)) {
+    return true;
+  }
+  return owner_thread_ == current_thread_id;
+}
+
+void Isolate::ReleaseOwnership() {
+  bool result = SetOwnerThread(OSThread::GetCurrentThreadId(),
+                               OSThread::kInvalidThreadId);
+  ASSERT(result);
+}
+
 FfiCallbackMetadata::Trampoline Isolate::CreateAsyncFfiCallback(
     Zone* zone,
     const Function& send_function,

runtime/vm/isolate.h

@@ -1294,6 +1294,14 @@ class Isolate : public IntrusiveDListEntry<Isolate> {
   void DecrementSpawnCount();
   void WaitForOutstandingSpawns();
 
+  bool TryAcquireOwnership();
+  void ReleaseOwnership();
+  bool is_permanently_pinned() { return is_permanently_pinned_; }
+  void set_is_permanently_pinned() { is_permanently_pinned_ = true; }
+  void clear_is_permanently_pinned_for_testing_only() {
+    is_permanently_pinned_ = false;
+  }
+
   static void SetCreateGroupCallback(Dart_IsolateGroupCreateCallback cb) {
     create_group_callback_ = cb;
   }
@@ -1700,6 +1708,7 @@ class Isolate : public IntrusiveDListEntry<Isolate> {
   FfiCallbackMetadata::MetadataEntry* ffi_callback_list_head_ = nullptr;
   intptr_t ffi_callback_keep_alive_counter_ = 0;
   RelaxedAtomic<ThreadId> owner_thread_ = OSThread::kInvalidThreadId;
+  bool is_permanently_pinned_ = false;
 
   ErrorPtr sticky_error_;
 

runtime/vm/message.h

@@ -58,7 +58,7 @@ class Message {
   // the VM heap. This is indicated by setting the len_ field to 0.
   Message(Dart_Port dest_port, ObjectPtr raw_obj, Priority priority);
 
-  // A message sent from SendPort.send or SendPort.sendAndExit where sender and
+  // A message sent from SendPort.send or Isolate.exit where sender and
   // receiver are in the same isolate group.
   Message(Dart_Port dest_port, PersistentHandle* handle, Priority priority);
 
@@ -109,7 +109,7 @@ class Message {
   }
   // A message whose object is an immortal object from the vm-isolate's heap.
   bool IsRaw() const { return snapshot_length_ == 0; }
-  // A message sent from SendPort.send or SendPort.sendAndExit where sender and
+  // A message sent from SendPort.send or Isolate.exit where sender and
   // receiver are in the same isolate group.
   bool IsPersistentHandle() const {
     return snapshot_length_ == kPersistentHandleSnapshotLen;

runtime/vm/message_handler.h

@@ -137,6 +137,8 @@ class MessageHandler : public PortHandler {
   void PostMessage(std::unique_ptr<Message> message,
                    bool before_events = false) override;
 
+  bool is_scheduled() { return pool_ != nullptr; }
+
  private:
   template <typename GCVisitorType>
   friend void MournFinalizerEntry(GCVisitorType*, FinalizerEntryPtr);

runtime/vm/port.cc

@@ -9,10 +9,12 @@
 #include "include/dart_api.h"
 #include "platform/utils.h"
 #include "vm/dart_api_impl.h"
+#include "vm/dart_api_message.h"
 #include "vm/dart_entry.h"
 #include "vm/isolate.h"
 #include "vm/lockers.h"
 #include "vm/message_handler.h"
+#include "vm/message_snapshot.h"
 #include "vm/os_thread.h"
 
 namespace dart {
@@ -206,6 +208,16 @@ Dart_Port PortMap::GetOriginId(Dart_Port id) {
   return isolate->group()->id();
 }
 
+bool PortMap::IsOwned(Dart_Port id) {
+  Locker ml;
+  Isolate* isolate = GetIsolateLocked(ml, id);
+  if (isolate == nullptr) {
+    // Either the port is invalid, or the isolate has already shut down.
+    return false;
+  }
+  return isolate->GetOwnerThread(&ml) != OSThread::kInvalidThreadId;
+}
+
 bool PortMap::IsOwnedByCurrentThread(Dart_Port id) {
   Locker ml;
   Isolate* isolate = GetIsolateLocked(ml, id);
@@ -216,6 +228,50 @@ bool PortMap::IsOwnedByCurrentThread(Dart_Port id) {
   return isolate->GetOwnerThread(&ml) == OSThread::GetCurrentThreadId();
 }
 
+IsolateAcquireResult PortMap::AcquireIsolateByControlPort(Dart_Port target_port,
+                                                          Isolate** p_isolate) {
+  ASSERT(p_isolate != nullptr);
+  Locker ml;  // isolates are not exiting while we hold this lock
+  if (ports_ == nullptr) {
+    return IsolateAcquireResult::ISOLATE_NOT_AVAILABLE;
+  }
+  auto it = ports_->TryLookup(target_port);
+  if (it == ports_->end()) {
+    return IsolateAcquireResult::ISOLATE_NOT_AVAILABLE;
+  }
+  auto target_handler = (*it).handler;
+  ASSERT(target_handler != nullptr);
+  auto target_isolate = target_handler->isolate();
+
+  if (!target_isolate->TryAcquireOwnership()) {
+    return target_isolate->is_permanently_pinned()
+               ? IsolateAcquireResult::PINNED_TO_ANOTHER_THREAD
+               : IsolateAcquireResult::BUSY;
+  }
+
+  *p_isolate = target_isolate;
+  return IsolateAcquireResult::SUCCESS;
+}
+
+bool PortMap::HasEventLoopRunning(Dart_Port id) {
+  Locker ml;  // isolates are not exiting while we hold this lock
+  if (ports_ == nullptr) {
+    return false;
+  }
+  auto it = ports_->TryLookup(id);
+  if (it == ports_->end()) {
+    return false;
+  }
+  auto target_handler = (*it).handler;
+  ASSERT(target_handler != nullptr);
+  auto isolate = target_handler->isolate();
+  if (isolate->message_notify_callback() != nullptr) {
+    return true;
+  }
+  auto message_handler = isolate->message_handler();
+  return message_handler != nullptr && message_handler->is_scheduled();
+}
+
 #if defined(TESTING)
 bool PortMap::HasPorts(MessageHandler* handler) {
   Locker ml;

runtime/vm/port.h

@@ -23,6 +23,13 @@ class MessageHandler;
 class Mutex;
 class PortHandler;
 
+enum class IsolateAcquireResult {
+  SUCCESS,
+  ISOLATE_NOT_AVAILABLE,
+  BUSY,
+  PINNED_TO_ANOTHER_THREAD,
+};
+
 class PortMap : public AllStatic {
  public:
   // Allocate a port for the provided handler and return its VM-global id.
@@ -50,6 +57,14 @@ class PortMap : public AllStatic {
   // thread.
   static bool IsOwnedByCurrentThread(Dart_Port id);
 
+  // Returns true if the port is owned by somebody.
+  static bool IsOwned(Dart_Port id);
+
+  static bool HasEventLoopRunning(Dart_Port id);
+
+  static IsolateAcquireResult AcquireIsolateByControlPort(Dart_Port target_port,
+                                                          Isolate** p_isolate);
+
 #if defined(TESTING)
   static Isolate* GetIsolate(Dart_Port id);
   static bool PortExists(Dart_Port id);

runtime/vm/thread.cc

@@ -490,6 +490,8 @@ void Thread::ExitIsolate(bool isolate_shutdown) {
   ASSERT(thread->isolate()->scheduled_mutator_thread_ == thread);
   DEBUG_ASSERT(!thread->IsAnyReusableHandleScopeActive());
 
+  ASSERT(thread->execution_state() == Thread::kThreadInVM);
+
   auto isolate = thread->isolate();
   auto group = thread->isolate_group();
 
@@ -563,15 +565,30 @@ void Thread::ExitIsolateGroupAsHelper(bool bypass_safepoint) {
 }
 
 void Thread::EnterIsolateGroupAsMutator(IsolateGroup* isolate_group,
-                                        bool bypass_safepoint) {
+                                        bool bypass_safepoint,
+                                        Thread* suspended_thread) {
   Roots::SetCurrent(isolate_group->roots());
   isolate_group->IncreaseMutatorCount(/*thread=*/nullptr,
                                       /*is_nested_reenter=*/true,
                                       /*was_stolen=*/false);
   isolate_group->IncrementIsolateGroupMutatorCount();
-  Thread* thread = AddActiveThread(isolate_group, /*isolate=*/nullptr,
-                                   kMutatorTask, bypass_safepoint);
 
+  auto thread = suspended_thread;
+  if (thread != nullptr) {
+    ResumeThreadInternal(thread);
+    {
+      // Descheduled isolates are reloadable (if nothing else prevents it).
+      RawReloadParticipationScope enable_reload(thread);
+      thread->ExitSafepoint();
+    }
+
+    thread->AssertDartMutatorInvariants();
+    ASSERT(thread->isolate() == nullptr);
+    ASSERT(thread->isolate_group() == isolate_group);
+    return;
+  }
+  thread = AddActiveThread(isolate_group, /*isolate=*/nullptr, kMutatorTask,
+                           bypass_safepoint);
   RELEASE_ASSERT(thread != nullptr);
   // Even if [bypass_safepoint] is true, a thread may need mutator state (e.g.
   // parallel scavenger threads write to the [Thread]s storebuffer)
@@ -607,15 +624,26 @@ void Thread::EnterIsolateGroupAsMutator(IsolateGroup* isolate_group,
 void Thread::ExitIsolateGroupAsMutator(bool bypass_safepoint) {
   Thread* thread = Thread::Current();
   thread->AssertDartMutatorInvariants();
+  auto group = thread->isolate_group();
 
   // Even if [bypass_safepoint] is true, a thread may need mutator state (e.g.
   // parallel scavenger threads write to the [Thread]s storebuffer)
-  thread->ResetDartMutatorState();
-  thread->ResetMutatorState();
-  thread->ClearStackLimit();
-  SuspendThreadInternal(thread, VMTag::kInvalidTagId);
-  auto group = thread->isolate_group();
-  FreeActiveThread(thread, /*isolate=*/nullptr, bypass_safepoint);
+  if (thread->HasActiveState() || thread->OwnsSafepoint()) {
+    // must not free the thread
+    SuspendThreadInternal(thread, VMTag::kLoadWaitTagId);
+    {
+      // Descheduled isolates are reloadable (if nothing else prevents it).
+      RawReloadParticipationScope enable_reload(thread);
+      thread->EnterSafepoint();
+    }
+    thread->set_execution_state(Thread::kThreadInNative);
+  } else {
+    thread->ResetDartMutatorState();
+    thread->ResetMutatorState();
+    thread->ClearStackLimit();
+    SuspendThreadInternal(thread, VMTag::kInvalidTagId);
+    FreeActiveThread(thread, /*isolate=*/nullptr, bypass_safepoint);
+  }
   group->DecrementIsolateGroupMutatorCount();
   group->DecreaseMutatorCount(/*is_nested_exit=*/true);
   Roots::ClearCurrent();

runtime/vm/thread.h

@@ -406,7 +406,8 @@ class Thread : public ThreadState, public IntrusiveDListEntry<Thread> {
   static void ExitIsolateGroupAsNonMutator();
 
   static void EnterIsolateGroupAsMutator(IsolateGroup* isolate_group,
-                                         bool bypass_safepoint);
+                                         bool bypass_safepoint,
+                                         Thread* suspended_thread = nullptr);
   static void ExitIsolateGroupAsMutator(bool bypass_safepoint);
 
   // Empties the store buffer block into the isolate.

sdk/lib/_internal/js_dev_runtime/patch/isolate_patch.dart

@@ -93,6 +93,30 @@ class Isolate {
   @patch
   static Never exit([SendPort? finalMessagePort, Object? message]) =>
       _unsupported();
+
+  @patch
+  static Isolate create({String? debugName}) => _unsupported();
+
+  @patch
+  void shutdownSync() => _unsupported();
+
+  @patch
+  R runSync<R>(R Function() f) => _unsupported();
+
+  @patch
+  static bool pinToCurrentThread() => _unsupported();
+
+  @patch
+  bool get isPinnedToCurrentThread => _unsupported();
+
+  @patch
+  void runEventLoopSync() => _unsupported();
+
+  @patch
+  void set onEvent(void Function(Isolate) callback) => _unsupported();
+
+  @patch
+  void handleEvent() => _unsupported();
 }
 
 /// Default factory for receive ports.

sdk/lib/_internal/js_runtime/lib/isolate_patch.dart

@@ -126,6 +126,46 @@ class Isolate {
   static Never exit([SendPort? finalMessagePort, Object? message]) {
     throw UnsupportedError("Isolate.exit");
   }
+
+  @patch
+  static Isolate create({String? debugName}) {
+    throw UnsupportedError("Isolate.create");
+  }
+
+  @patch
+  void shutdownSync() {
+    throw UnsupportedError("Isolate.shutdownSync");
+  }
+
+  @patch
+  R runSync<R>(R Function() f) {
+    throw UnsupportedError("Isolate.runSync");
+  }
+
+  @patch
+  static bool pinToCurrentThread() {
+    throw UnsupportedError("Isolate.pintToCurrentThread");
+  }
+
+  @patch
+  bool get isPinnedToCurrentThread {
+    throw UnsupportedError("Isolate.isPinnedToCurrentThread");
+  }
+
+  @patch
+  void runEventLoopSync() {
+    throw UnsupportedError("Isolate.runEventLoopSync");
+  }
+
+  @patch
+  void set onEvent(void Function(Isolate) callback) {
+    throw UnsupportedError("Isolate.onEvent");
+  }
+
+  @patch
+  void handleEvent() {
+    throw UnsupportedError("Isolate.handleEvent");
+  }
 }
 
 @patch

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

@@ -731,6 +731,62 @@ final class Isolate {
     }
     _exit(finalMessagePort, message);
   }
+
+  @patch
+  static Isolate create({String? debugName}) {
+    final List created = _create(debugName);
+    final SendPort controlPort = created[0];
+    final List capabilities = created[1];
+    return Isolate(
+      controlPort,
+      pauseCapability: capabilities[0],
+      terminateCapability: capabilities[1],
+    );
+  }
+
+  @pragma("vm:external-name", "Isolate_create_")
+  external static List _create(String? debugName);
+
+  @patch
+  void shutdownSync() {
+    _shutdownSync(controlPort);
+  }
+
+  @pragma("vm:external-name", "Isolate_shutdownSync_")
+  external static void _shutdownSync(SendPort controlPort);
+
+  @patch
+  R runSync<R>(R Function() f) {
+    return _runSync(controlPort, f);
+  }
+
+  @pragma("vm:external-name", "Isolate_runSync_")
+  external static R _runSync<R>(SendPort controlPort, R Function() f);
+
+  @patch
+  void runEventLoopSync() {
+    throw UnsupportedError("Isolate.runEventLoopSync");
+  }
+
+  @patch
+  static bool pinToCurrentThread() {
+    throw UnsupportedError("Isolate.pintToCurrentThread");
+  }
+
+  @patch
+  bool get isPinnedToCurrentThread {
+    throw UnsupportedError("Isolate.isPinnedToCurrentThread");
+  }
+
+  @patch
+  void set onEvent(void Function(Isolate) callback) {
+    throw UnsupportedError("Isolate.onEvent");
+  }
+
+  @patch
+  void handleEvent() {
+    throw UnsupportedError("Isolate.handleEvent");
+  }
 }
 
 @patch

sdk/lib/_internal/wasm/common/isolate_patch.dart

@@ -125,6 +125,46 @@ class Isolate {
   static Never exit([SendPort? finalMessagePort, Object? message]) {
     throw UnsupportedError("Isolate.exit");
   }
+
+  @patch
+  static Isolate create({String? debugName}) {
+    throw UnsupportedError("Isolate.create");
+  }
+
+  @patch
+  void shutdownSync() {
+    throw UnsupportedError("Isolate.shutdownSync");
+  }
+
+  @patch
+  R runSync<R>(R Function() f) {
+    throw UnsupportedError("Isolate.runSync");
+  }
+
+  @patch
+  static bool pinToCurrentThread() {
+    throw UnsupportedError("Isolate.pintToCurrentThread");
+  }
+
+  @patch
+  bool get isPinnedToCurrentThread {
+    throw UnsupportedError("Isolate.isPinnedToCurrentThread");
+  }
+
+  @patch
+  void runEventLoopSync() {
+    throw UnsupportedError("Isolate.runEventLoopSync");
+  }
+
+  @patch
+  void set onEvent(void Function(Isolate) callback) {
+    throw UnsupportedError("Isolate.onEvent");
+  }
+
+  @patch
+  void handleEvent() {
+    throw UnsupportedError("Isolate.handleEvent");
+  }
 }
 
 @patch

sdk/lib/isolate/isolate.dart

@@ -832,6 +832,124 @@ final class Isolate {
   /// receiving isolate will in most cases be able to receive the message
   /// in constant time.
   external static Never exit([SendPort? finalMessagePort, Object? message]);
+
+  /// Execute the given function in the context of the given isolate.
+  ///
+  /// This function will throw if target isolate is running.
+  ///
+  /// Throws an error if target isolate is pinned to another thread and
+  /// thus can't be entered from this thread. See [pinToCurrentThread] and
+  /// [isPinnedToCurrentThread].
+  ///
+  /// Throws an error if the target isolate belongs to another isolate group.
+  ///
+  /// Throws an error if [f] is not deeply immutable.
+  ///
+  /// Throws an error if result returned by [f] is not deeply immutable.
+  @Since("3.13")
+  external R runSync<R>(R Function() f);
+
+  /// Create a new isolate in the current isolate group.
+  ///
+  /// Similar to `Dart_CreateIsolateInGroup` Dart VM C API.
+  ///
+  /// The isolate has been created, but its event loop is not running.
+  ///
+  /// To start processing isolate's messages:
+  ///
+  /// * start isolate's event loop synchronously on the current thread
+  ///   by calling [Isolate.runEventLoopSync]
+  /// * integrate isolate's event loop with an external event loop by
+  ///   registering event callback ([Isolate.onEvent]) to forward
+  ///   event notifications to an external event loop and then draining
+  ///   pending events ([Isolate.handleEvent]) from that event loop.
+  @Since("3.13")
+  external static Isolate create({String? debugName});
+
+  /// Shut down target isolate.
+  ///
+  /// Shutting down the isolate stops its event loop without processing
+  /// any pending messages and closes all open receive ports owned by the
+  /// isolate.
+  ///
+  /// This function will block until it acquires exclusive access to the
+  /// target isolate. Isolate can only be entered for synchronous execution
+  /// between turns of its event loop, when no other thread is
+  /// executing code in the target isolate.
+  @Since("3.13")
+  external void shutdownSync();
+
+  /// Pin current isolate to the current OS thread.
+  ///
+  /// Once an isolate is pinned to an OS thread it cannot be
+  /// entered by any other OS thread. An attempt to acquire
+  /// exclusive access to it from another thread will fail with
+  /// an error.
+  ///
+  /// Equivalent to `Dart_SetCurrentThreadOwnsIsolate` Dart VM C API.
+  ///
+  /// Returns `true` on success and `false` otherwise (e.g. if target isolate
+  /// is already pinned to another thread).
+  @Since("3.13")
+  external static bool pinToCurrentThread();
+
+  /// Whether the isolate is pinned to the current OS thread.
+  ///
+  /// Equivalent to `Dart_GetCurrentThreadOwnsIsolate` Dart VM C API.
+  @Since("3.13")
+  external bool get isPinnedToCurrentThread;
+
+  /// Run event loop for the target isolate synchronously on the current thread.
+  ///
+  /// This function will block until it acquires exclusive access to the
+  /// target isolate. Isolate can only be entered for synchronous execution
+  /// between turns of its event loop, when no other thread is
+  /// executing code in the target isolate.
+  ///
+  /// This function will return once the isolate has no open keep-alive
+  /// receive ports.
+  ///
+  /// The isolate will be marked as pinned to the current thread.
+  ///
+  /// Similar to `Dart_RunLoop` Dart VM C API, but unlike `Dart_RunLoop` this
+  /// function executes isolate's event loop on the current thread instead
+  /// of delegating it into the thread-pool.
+  ///
+  /// Throws an error if target isolate is pinned to another thread or already
+  /// has an event loop running.
+  @Since("3.13")
+  external void runEventLoopSync();
+
+  /// Event notify callback for the isolate.
+  ///
+  /// Provided callback will be called once for every new event which isolate
+  /// needs to react to. Pending events can be then later be drained
+  /// by calling [Isolate.handleEvent].
+  ///
+  /// Provided [callback] must be deeply immutable and will be called
+  /// on an arbitrary thread and not necessarily within any isolate. See
+  /// [NativeCallable.isolateGroupBound].
+  ///
+  /// IMPORTANT: [Isolate.handleEvent] *MUST NOT* be called from the
+  /// `callback` as this will cause a dead-locks of the Dart execution
+  /// environment.
+  ///
+  /// Similar to `Dart_SetMessageNotifyCallback` Dart VM C API.
+  @Since("3.13")
+  external void set onEvent(void Function(Isolate) callback);
+
+  /// Handle at most one pending event for the isolate.
+  ///
+  /// This function does nothing if there are no pending events.
+  ///
+  /// This function will block until it acquires exclusive access to the
+  /// target isolate. Isolate can only be entered for synchronous execution
+  /// between turns of its event loop, when no other thread is
+  /// executing code in the target isolate.
+  ///
+  /// Similar to `Dart_HandleMessage` Dart VM C API.
+  @Since("3.13")
+  external void handleEvent();
 }
 
 /// Sends messages to its [ReceivePort]s.

tests/ffi/threading_test.dart

@@ -0,0 +1,398 @@
+// 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 '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');
+
+@pragma('vm:shared')
+int counter = 0;
+
+void testRunSyncOnCurrentIsolate() {
+  // Run on current isolate.
+  final current_isolate = Isolate.current;
+  Expect.equals(
+    42,
+    current_isolate.runSync(() {
+      Expect.equals(
+        56,
+        Isolate.current.runSync(() {
+          return 56;
+        }),
+      );
+      return 42;
+    }),
+  );
+}
+
+Future<void> testFailRunSyncOnAnotherIsolate() async {
+  final rpChild = ReceivePort();
+  final rpChildExit = ReceivePort();
+  final child = await Isolate.spawn(
+    (sendPort) async {
+      final rp = ReceivePort();
+      sendPort.send(rp.sendPort);
+      await rp.first;
+      rp.close();
+    },
+    rpChild.sendPort,
+    onExit: rpChildExit.sendPort,
+  );
+  SendPort rpChildRequestExit = await rpChild.first;
+  Expect.throws(
+    () => child.runSync(() {}),
+    (e) => e is StateError && e.message.contains("Isolate has a message loop"),
+  );
+  rpChildRequestExit.send(true);
+  await rpChildExit.first;
+  rpChildExit.close();
+  rpChild.close();
+}
+
+void testRunSyncChecks() {
+  final isolate = Isolate.current;
+  // Only deeply immutable values can be returned from runSync closure.
+  Expect.throwsArgumentError(() {
+    isolate.runSync(() => RawReceivePort()..keepIsolateAlive = false);
+  });
+  // SendPort can be returned from runSync closure.
+  Expect.isNotNull(
+    isolate.runSync(() {
+      final rrp = RawReceivePort()..keepIsolateAlive = false;
+      return rrp.sendPort;
+    }),
+  );
+
+  // Only deeply immutable values can be captured by runSync closure.
+  {
+    final rp = RawReceivePort();
+    Expect.throwsArgumentError(() {
+      isolate.runSync(() => rp.sendPort);
+    });
+    rp.close();
+  }
+}
+
+Future<void> testFailToRunOnExitedIsolate() async {
+  // Enter isolate that never gets to the finish single message loop iteration.
+  counter = 0;
+  final rp = ReceivePort();
+  final rpChildExit = ReceivePort();
+  final child = await Isolate.spawn(
+    (sendPort) async {
+      final rpChildListening = ReceivePort();
+      sendPort.send(rpChildListening.sendPort);
+      await rpChildListening.first;
+    },
+    rp.sendPort,
+    onExit: rpChildExit.sendPort,
+  );
+  final spChildListening = await rp.first;
+  Expect.throws(
+    () => child.runSync(() {
+      print('child runSync is running');
+    }),
+    (e) =>
+        e is StateError &&
+        e.message.contains("Isolate has a message loop running"),
+  );
+  spChildListening.send('you can exit now');
+  await rpChildExit.first;
+  rpChildExit.close();
+  Expect.throws(
+    () => child.runSync(() {
+      print('child runSync is running');
+    }),
+    (e) => e is StateError && e.message.contains("Unable to enter the isolate"),
+  );
+  rp.close();
+}
+
+@pragma('vm:shared')
+final dartSetCurrentThreadOwnsIsolate = DynamicLibrary.executable()
+    .lookup<NativeFunction<Void Function()>>("Dart_SetCurrentThreadOwnsIsolate")
+    .asFunction<void Function()>();
+
+int threadMain(Pointer<Void> data) {
+  final new_isolate = Isolate.create(debugName: "helper");
+  new_isolate.runSync(() {
+    dartSetCurrentThreadOwnsIsolate();
+  });
+  new_isolate.runSync(() {
+    print('Hello, new isolate!');
+  });
+  new_isolate.shutdownSync();
+  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.
+  }
+
+  final callback =
+      NativeCallable<IntPtr Function(Pointer<Void>)>.isolateGroupBound(
+        threadMain,
+        exceptionalReturn: -1,
+      );
+
+  final threadInfo = ThreadInfo();
+  Expect.equals(0, pthreadAttrInit(threadInfo.ptr_attr));
+  Expect.equals(
+    0,
+    pthreadCreate(
+      threadInfo.ptr_tid,
+      threadInfo.ptr_attr,
+      callback.nativeFunction,
+      threadInfo.ptr_data.cast<Void>(),
+    ),
+  );
+
+  threadInfo.join();
+}
+
+@pragma('vm:shared')
+late SendPort sp;
+@pragma('vm:shared')
+final Mutex mutexCondvar = Mutex();
+@pragma('vm:shared')
+final ConditionVariable condVar = ConditionVariable();
+@pragma('vm:shared')
+bool latchOpened = false;
+
+void waitLatch() {
+  mutexCondvar.runLocked(() {
+    while (!latchOpened) {
+      condVar.wait(mutexCondvar);
+    }
+    latchOpened = false;
+  });
+}
+
+void openLatch() {
+  mutexCondvar.runLocked(() {
+    latchOpened = true;
+    condVar.notify();
+  });
+}
+
+int threadMainPinned(Pointer<Void> data) {
+  final new_isolate = Isolate.create(debugName: "helper");
+
+  new_isolate.runSync(() {
+    dartSetCurrentThreadOwnsIsolate();
+  });
+  new_isolate.runSync(() {
+    print('Hello, new pinned isolate!');
+  });
+  sp.send(new_isolate);
+  waitLatch();
+
+  new_isolate.shutdownSync();
+  return 0;
+}
+
+Future<void> testFailRunSyncOnPinnedIsolate() async {
+  if (Platform.isWindows) {
+    return; // pthread is not available on Windows.
+  }
+
+  final completer = Completer();
+  final rp = RawReceivePort((Isolate child_isolate) {
+    print('received $child_isolate');
+    Expect.throws(
+      () => child_isolate.runSync(() {
+        Expect.fail("Should not run");
+      }),
+      (e) =>
+          e is StateError &&
+          e.message.contains("Isolate is pinned to a different thread already"),
+    );
+    openLatch();
+    completer.complete();
+  });
+  sp = rp.sendPort;
+
+  final callback =
+      NativeCallable<IntPtr Function(Pointer<Void>)>.isolateGroupBound(
+        threadMainPinned,
+        exceptionalReturn: -1,
+      );
+  final threadInfo = ThreadInfo();
+  Expect.equals(0, pthreadAttrInit(threadInfo.ptr_attr));
+  Expect.equals(
+    0,
+    pthreadCreate(
+      threadInfo.ptr_tid,
+      threadInfo.ptr_attr,
+      callback.nativeFunction,
+      threadInfo.ptr_data.cast<Void>(),
+    ),
+  );
+
+  await completer.future;
+  rp.close();
+
+  threadInfo.join();
+}
+
+int threadMainWaitingLatch(Pointer<Void> data) {
+  final helper = Isolate.create(debugName: "helper");
+
+  sp.send(helper);
+  helper.runSync(() {
+    waitLatch();
+  });
+  print('shutting down the isolate');
+  helper.shutdownSync();
+  return 0;
+}
+
+Future<void> testFailRunSyncWithTimeout() async {
+  if (Platform.isWindows) {
+    return; // pthread is not available on Windows.
+  }
+
+  final completer = Completer();
+  final rp = RawReceivePort((Isolate child_isolate) {
+    print('received $child_isolate');
+    Expect.throws(
+      () => child_isolate.runSync(() {
+        Expect.fail("Should not run");
+      }),
+      (e) =>
+          e is StateError &&
+          e.message.contains("Isolate is busy, running on a different thread"),
+    );
+    openLatch();
+    completer.complete();
+  });
+  sp = rp.sendPort;
+
+  final callback =
+      NativeCallable<IntPtr Function(Pointer<Void>)>.isolateGroupBound(
+        threadMainWaitingLatch,
+        exceptionalReturn: -1,
+      );
+  final threadInfo = ThreadInfo();
+  Expect.equals(0, pthreadAttrInit(threadInfo.ptr_attr));
+  Expect.equals(
+    0,
+    pthreadCreate(
+      threadInfo.ptr_tid,
+      threadInfo.ptr_attr,
+      callback.nativeFunction,
+      threadInfo.ptr_data.cast<Void>(),
+    ),
+  );
+
+  await completer.future;
+  rp.close();
+
+  threadInfo.join();
+}
+
+Future<void> testFailRunSyncDifferentIsolateGroup() async {
+  final isolate = await Isolate.spawnUri(Platform.script, <String>[
+    "worker",
+  ], null);
+  Expect.isNotNull(isolate);
+  Expect.throws(
+    () => isolate.runSync(() {
+      Expect.fail("should not run");
+    }),
+    (e) =>
+        e is StateError &&
+        e.message.contains(
+          "Target isolate should be part of the same isolate group.",
+        ),
+  );
+}
+
+main(List<String> args, List<SendPort>? message) async {
+  if (message != null) {
+    Expect.equals(1, args.length);
+    Expect.equals("worker", args[0]);
+    await ReceivePort().first;
+    return;
+  }
+
+  asyncStart();
+
+  final isolates = List<Future<bool>>.generate(
+    30,
+    (i) => Isolate.run(() async {
+      testRunSyncOnCurrentIsolate();
+      await testFailRunSyncOnAnotherIsolate();
+
+      testRunSyncChecks();
+
+      await testFailToRunOnExitedIsolate();
+      return true;
+    }, debugName: 'worker isolate $i'),
+  );
+  await Future.wait(isolates);
+
+  await testRunSyncOnPinnedToSelfIsolate();
+  await testFailRunSyncOnPinnedIsolate();
+  await testFailRunSyncWithTimeout();
+  await testFailRunSyncDifferentIsolateGroup();
+
+  asyncEnd();
+}