diff --git a/runtime/lib/isolate.cc b/runtime/lib/isolate.cc index 8791e858ab7..1b084bd40e6 100644 --- a/runtime/lib/isolate.cc +++ b/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(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(); } diff --git a/runtime/vm/bootstrap_natives.h b/runtime/vm/bootstrap_natives.h index 8d42eaf3a5d..2fa4f13d3ea 100644 --- a/runtime/vm/bootstrap_natives.h +++ b/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) \ diff --git a/runtime/vm/message_handler.cc b/runtime/vm/message_handler.cc index 0afda79ac69..41b114d3885 100644 --- a/runtime/vm/message_handler.cc +++ b/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, bool before_events) { Message::Priority saved_priority; diff --git a/runtime/vm/message_handler.h b/runtime/vm/message_handler.h index 20a406e82c9..b45691f8744 100644 --- a/runtime/vm/message_handler.h +++ b/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). diff --git a/sdk/lib/_internal/vm/lib/isolate_patch.dart b/sdk/lib/_internal/vm/lib/isolate_patch.dart index 754fd0f4ae2..5283d9c1740 100644 --- a/sdk/lib/_internal/vm/lib/isolate_patch.dart +++ b/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"); diff --git a/tests/ffi/threading_runeventloop_test.dart b/tests/ffi/threading_runeventloop_test.dart new file mode 100644 index 00000000000..5d3b041b59d --- /dev/null +++ b/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 data) { + final pthreadSelf = DynamicLibrary.process() + .lookupFunction('pthread_self'); + final self = pthreadSelf(); + final i = data.cast()[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('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) threadMain, +) { + final threadInfo = ThreadInfo(); + + Expect.equals(0, pthreadAttrInit(threadInfo.ptr_attr)); + threadInfo.ptr_data.cast()[0] = i; + print( + '=== ptr_data: ${threadInfo.ptr_data.address.toRadixString(16)}, i: $i', + ); + final callback = + NativeCallable)>.isolateGroupBound( + threadMain, + exceptionalReturn: -1, + ); + callback.keepIsolateAlive = false; + pthreadCreate( + threadInfo.ptr_tid, + threadInfo.ptr_attr, + callback.nativeFunction, + threadInfo.ptr_data.cast(), + ); + return threadInfo; +} + +Future testRunEventLoopManyThreads({int numThreads = 63}) async { + if (Platform.isWindows) { + // pthread library loading doesn't work on Windows. + return; + } + mutexCondvar = Mutex(); + condVar = ConditionVariable(); + final threadInfos = []; + 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 testFailRunEventLoopFromIsolate() async { + Expect.throws( + () { + Isolate.current.runEventLoopSync(); + }, + (e) => + e is StateError && + e.message.contains("Should be invoked outside of an isolate"), + ); +} + +main(List 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(); +} diff --git a/tests/ffi/threading_test.dart b/tests/ffi/threading_test.dart index 1bd3a2c7bed..7720b6d5394 100644 --- a/tests/ffi/threading_test.dart +++ b/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); -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'); +import 'threading_utils.dart'; @pragma('vm:shared') int counter = 0; @@ -163,24 +140,6 @@ int threadMain(Pointer data) { 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. diff --git a/tests/ffi/threading_utils.dart b/tests/ffi/threading_utils.dart new file mode 100644 index 00000000000..73fd5706de0 --- /dev/null +++ b/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); +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'); + +typedef PthreadSelfFT = int Function(); +typedef PthreadSelfNFT = IntPtr Function(); +final pthreadSelf = DynamicLibrary.process() + .lookupFunction('pthread_self'); + +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); + } +}