diff --git a/runtime/lib/isolate.cc b/runtime/lib/isolate.cc index d4191b40570..8791e858ab7 100644 --- a/runtime/lib/isolate.cc +++ b/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( diff --git a/runtime/vm/bootstrap_natives.h b/runtime/vm/bootstrap_natives.h index 11f8512c9f1..8d42eaf3a5d 100644 --- a/runtime/vm/bootstrap_natives.h +++ b/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) \ diff --git a/runtime/vm/dart_api_impl.cc b/runtime/vm/dart_api_impl.cc index d223f456adc..32b613d73b6 100644 --- a/runtime/vm/dart_api_impl.cc +++ b/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) { diff --git a/runtime/vm/isolate.cc b/runtime/vm/isolate.cc index a430745f98d..a3ff0aa8947 100644 --- a/runtime/vm/isolate.cc +++ b/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, diff --git a/runtime/vm/isolate.h b/runtime/vm/isolate.h index 7a1ef5971ab..9a8d77567a6 100644 --- a/runtime/vm/isolate.h +++ b/runtime/vm/isolate.h @@ -1294,6 +1294,14 @@ class Isolate : public IntrusiveDListEntry { 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 { FfiCallbackMetadata::MetadataEntry* ffi_callback_list_head_ = nullptr; intptr_t ffi_callback_keep_alive_counter_ = 0; RelaxedAtomic owner_thread_ = OSThread::kInvalidThreadId; + bool is_permanently_pinned_ = false; ErrorPtr sticky_error_; diff --git a/runtime/vm/message.h b/runtime/vm/message.h index b10b16f7541..b614fe69a95 100644 --- a/runtime/vm/message.h +++ b/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; diff --git a/runtime/vm/message_handler.h b/runtime/vm/message_handler.h index 1021af328d7..20a406e82c9 100644 --- a/runtime/vm/message_handler.h +++ b/runtime/vm/message_handler.h @@ -137,6 +137,8 @@ class MessageHandler : public PortHandler { void PostMessage(std::unique_ptr message, bool before_events = false) override; + bool is_scheduled() { return pool_ != nullptr; } + private: template friend void MournFinalizerEntry(GCVisitorType*, FinalizerEntryPtr); diff --git a/runtime/vm/port.cc b/runtime/vm/port.cc index 115d7b0c764..3ffe35d0aa0 100644 --- a/runtime/vm/port.cc +++ b/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; diff --git a/runtime/vm/port.h b/runtime/vm/port.h index 449da208d63..23e74edb0e5 100644 --- a/runtime/vm/port.h +++ b/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); diff --git a/runtime/vm/thread.cc b/runtime/vm/thread.cc index 0621e147621..887450d21cf 100644 --- a/runtime/vm/thread.cc +++ b/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(); diff --git a/runtime/vm/thread.h b/runtime/vm/thread.h index 0643cd51a65..dbc055b1801 100644 --- a/runtime/vm/thread.h +++ b/runtime/vm/thread.h @@ -406,7 +406,8 @@ class Thread : public ThreadState, public IntrusiveDListEntry { 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. diff --git a/sdk/lib/_internal/js_dev_runtime/patch/isolate_patch.dart b/sdk/lib/_internal/js_dev_runtime/patch/isolate_patch.dart index 7073bb63416..37b0ccc74d1 100644 --- a/sdk/lib/_internal/js_dev_runtime/patch/isolate_patch.dart +++ b/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 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. diff --git a/sdk/lib/_internal/js_runtime/lib/isolate_patch.dart b/sdk/lib/_internal/js_runtime/lib/isolate_patch.dart index 192971d2c9d..685892810d2 100644 --- a/sdk/lib/_internal/js_runtime/lib/isolate_patch.dart +++ b/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 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 diff --git a/sdk/lib/_internal/vm/lib/isolate_patch.dart b/sdk/lib/_internal/vm/lib/isolate_patch.dart index 7822310d79e..754fd0f4ae2 100644 --- a/sdk/lib/_internal/vm/lib/isolate_patch.dart +++ b/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 Function() f) { + return _runSync(controlPort, f); + } + + @pragma("vm:external-name", "Isolate_runSync_") + external static R _runSync(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 diff --git a/sdk/lib/_internal/wasm/common/isolate_patch.dart b/sdk/lib/_internal/wasm/common/isolate_patch.dart index ce678017708..3452be402cd 100644 --- a/sdk/lib/_internal/wasm/common/isolate_patch.dart +++ b/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 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 diff --git a/sdk/lib/isolate/isolate.dart b/sdk/lib/isolate/isolate.dart index 8e4cb2b6e6a..72349525852 100644 --- a/sdk/lib/isolate/isolate.dart +++ b/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 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. diff --git a/tests/ffi/threading_test.dart b/tests/ffi/threading_test.dart new file mode 100644 index 00000000000..1bd3a2c7bed --- /dev/null +++ b/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); +typedef PthreadAttrInitNFT = IntPtr Function(Pointer); +final pthreadAttrInit = DynamicLibrary.process() + .lookupFunction('pthread_attr_init'); + +typedef PthreadAttrDestroyFT = int Function(Pointer); +typedef PthreadAttrDestroyNFT = IntPtr Function(Pointer); +final pthreadAttrDestroy = DynamicLibrary.process() + .lookupFunction( + 'pthread_attr_destroy', + ); + +typedef PthreadCreateFT = + int Function(Pointer, Pointer, Pointer, Pointer); +typedef PthreadCreateNFT = + IntPtr Function(Pointer, Pointer, Pointer, Pointer); +final pthreadCreate = DynamicLibrary.process() + .lookupFunction('pthread_create'); + +typedef PthreadJoinFT = int Function(int, Pointer); +typedef PthreadJoinNFT = IntPtr Function(IntPtr, Pointer); +final pthreadJoin = DynamicLibrary.process() + .lookupFunction('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 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 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>("Dart_SetCurrentThreadOwnsIsolate") + .asFunction(); + +int threadMain(Pointer 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(64); // big enough to fit pthread_attr_t? + final ptr_tid = calloc(1); + final ptr_data = calloc(1024); + final ptr_retval = calloc(1024); + + void join() { + Expect.equals(0, pthreadJoin(ptr_tid.value, ptr_retval.cast())); + calloc.free(ptr_retval); + + calloc.free(ptr_data); + calloc.free(ptr_tid); + + Expect.equals(0, pthreadAttrDestroy(ptr_attr)); + calloc.free(ptr_attr); + } +} + +Future testRunSyncOnPinnedToSelfIsolate() async { + if (Platform.isWindows) { + return; // pthread is not available on Windows. + } + + final callback = + NativeCallable)>.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(), + ), + ); + + 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 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 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)>.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(), + ), + ); + + await completer.future; + rp.close(); + + threadInfo.join(); +} + +int threadMainWaitingLatch(Pointer data) { + final helper = Isolate.create(debugName: "helper"); + + sp.send(helper); + helper.runSync(() { + waitLatch(); + }); + print('shutting down the isolate'); + helper.shutdownSync(); + return 0; +} + +Future 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)>.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(), + ), + ); + + await completer.future; + rp.close(); + + threadInfo.join(); +} + +Future testFailRunSyncDifferentIsolateGroup() async { + final isolate = await Isolate.spawnUri(Platform.script, [ + "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 args, List? message) async { + if (message != null) { + Expect.equals(1, args.length); + Expect.equals("worker", args[0]); + await ReceivePort().first; + return; + } + + asyncStart(); + + final isolates = List>.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(); +}