From 02b30d5275fae8ef546db74d3258f5f61cda3fd4 Mon Sep 17 00:00:00 2001 From: Alexander Aprelev Date: Fri, 29 May 2026 07:58:23 -0700 Subject: [PATCH] [vm/shared] Implement Isolate pinToCurrentThread, isPinnedToCurrentThread. TEST=threading_pinning_test Change-Id: I9a80543a06dbf51c070fed5c0e64eeba247497de Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/497126 Reviewed-by: Slava Egorov Commit-Queue: Alexander Aprelev --- runtime/lib/isolate.cc | 31 ++++++ runtime/vm/bootstrap_natives.h | 2 + runtime/vm/dart_api_impl.cc | 2 +- runtime/vm/port.cc | 5 +- runtime/vm/port.h | 3 +- sdk/lib/_internal/vm/lib/isolate_patch.dart | 12 ++- tests/ffi/threading_pinning_test.dart | 105 ++++++++++++++++++++ 7 files changed, 153 insertions(+), 7 deletions(-) create mode 100644 tests/ffi/threading_pinning_test.dart diff --git a/runtime/lib/isolate.cc b/runtime/lib/isolate.cc index 1b084bd40e6..62d67a46f42 100644 --- a/runtime/lib/isolate.cc +++ b/runtime/lib/isolate.cc @@ -775,6 +775,37 @@ DEFINE_NATIVE_ENTRY(Isolate_runSync_, 1, 2) { return result.ptr(); } +DEFINE_NATIVE_ENTRY(Isolate_isPinnedToCurrentThread, 0, 1) { + GET_NON_NULL_NATIVE_ARGUMENT(SendPort, isolate_control_port, + arguments->NativeArgAt(0)); + auto control_port_id = isolate_control_port.Id(); + return Bool::Get(Dart_GetCurrentThreadOwnsIsolate(control_port_id)).ptr(); +} + +DEFINE_NATIVE_ENTRY(Isolate_pinToCurrentThread, 0, 0) { + if (isolate == nullptr) { + ThrowCantRunWithoutIsolateError(); + UNREACHABLE(); + } + CHECK_ISOLATE(isolate); + if (!isolate->SetOwnerThread(OSThread::kInvalidThreadId, + OSThread::GetCurrentThreadId())) { + // We might be running this method while running dart code + // on this target isolate. + // So first confirm that the isolate is not yet pinned yet. + if (isolate->is_permanently_pinned()) { + return Bool::False().ptr(); + } + // Allow pinning only if current owner is the current thread. + if (isolate->GetOwnerThread(/*locker=*/nullptr) != + OSThread::GetCurrentThreadId()) { + return Bool::False().ptr(); + } + } + isolate->set_is_permanently_pinned(); + return Bool::True().ptr(); +} + // TODO(http://dartbug.com/47777): Add support for Finalizers. DEFINE_NATIVE_ENTRY(Isolate_exit_, 0, 2) { if (isolate == nullptr) { diff --git a/runtime/vm/bootstrap_natives.h b/runtime/vm/bootstrap_natives.h index 67dee2f39c2..40756aab3e0 100644 --- a/runtime/vm/bootstrap_natives.h +++ b/runtime/vm/bootstrap_natives.h @@ -255,6 +255,8 @@ namespace dart { V(Isolate_getCurrentRootUriStr, 0) \ V(Isolate_getDebugName, 1) \ V(Isolate_getPortAndCapabilitiesOfCurrentIsolate, 0) \ + V(Isolate_isPinnedToCurrentThread, 1) \ + V(Isolate_pinToCurrentThread, 0) \ V(Isolate_runSync_, 2) \ V(Isolate_runEventLoopSync_, 1) \ V(Isolate_sendOOB, 2) \ diff --git a/runtime/vm/dart_api_impl.cc b/runtime/vm/dart_api_impl.cc index 32b613d73b6..7414d544730 100644 --- a/runtime/vm/dart_api_impl.cc +++ b/runtime/vm/dart_api_impl.cc @@ -2197,7 +2197,7 @@ DART_EXPORT void Dart_ClearCurrentThreadOwnsIsolate_ForTesting() { } DART_EXPORT bool Dart_GetCurrentThreadOwnsIsolate(Dart_Port port) { - return PortMap::IsOwnedByCurrentThread(port); + return PortMap::IsOwnedByCurrentThread(port, /*require_permanent_pin=*/true); } // --- Scopes ---- diff --git a/runtime/vm/port.cc b/runtime/vm/port.cc index 3ffe35d0aa0..c2f403e6c38 100644 --- a/runtime/vm/port.cc +++ b/runtime/vm/port.cc @@ -218,14 +218,15 @@ bool PortMap::IsOwned(Dart_Port id) { return isolate->GetOwnerThread(&ml) != OSThread::kInvalidThreadId; } -bool PortMap::IsOwnedByCurrentThread(Dart_Port id) { +bool PortMap::IsOwnedByCurrentThread(Dart_Port id, bool require_permanent_pin) { 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::GetCurrentThreadId(); + return isolate->GetOwnerThread(&ml) == OSThread::GetCurrentThreadId() && + (!require_permanent_pin || isolate->is_permanently_pinned()); } IsolateAcquireResult PortMap::AcquireIsolateByControlPort(Dart_Port target_port, diff --git a/runtime/vm/port.h b/runtime/vm/port.h index 23e74edb0e5..d7a7ba980e5 100644 --- a/runtime/vm/port.h +++ b/runtime/vm/port.h @@ -55,7 +55,8 @@ class PortMap : public AllStatic { // Returns whether the isolate that owns the port is owned by the current // thread. - static bool IsOwnedByCurrentThread(Dart_Port id); + static bool IsOwnedByCurrentThread(Dart_Port id, + bool require_permanent_pin = false); // Returns true if the port is owned by somebody. static bool IsOwned(Dart_Port id); diff --git a/sdk/lib/_internal/vm/lib/isolate_patch.dart b/sdk/lib/_internal/vm/lib/isolate_patch.dart index 5283d9c1740..520c5be90ef 100644 --- a/sdk/lib/_internal/vm/lib/isolate_patch.dart +++ b/sdk/lib/_internal/vm/lib/isolate_patch.dart @@ -749,7 +749,7 @@ final class Isolate { @patch void shutdownSync() { - _shutdownSync(controlPort); + Isolate._shutdownSync(controlPort); } @pragma("vm:external-name", "Isolate_shutdownSync_") @@ -768,14 +768,20 @@ final class Isolate { _runEventLoopSync(controlPort); } + @pragma("vm:external-name", "Isolate_pinToCurrentThread") + external static bool _pinToCurrentThread(); + @patch static bool pinToCurrentThread() { - throw UnsupportedError("Isolate.pintToCurrentThread"); + return _pinToCurrentThread(); } + @pragma("vm:external-name", "Isolate_isPinnedToCurrentThread") + external static bool _isPinnedToCurrentThread(SendPort controlPort); + @patch bool get isPinnedToCurrentThread { - throw UnsupportedError("Isolate.isPinnedToCurrentThread"); + return Isolate._isPinnedToCurrentThread(controlPort); } @pragma("vm:external-name", "Isolate_runEventLoopSync_") diff --git a/tests/ffi/threading_pinning_test.dart b/tests/ffi/threading_pinning_test.dart new file mode 100644 index 00000000000..a3b7a918708 --- /dev/null +++ b/tests/ffi/threading_pinning_test.dart @@ -0,0 +1,105 @@ +// 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:expect/async_helper.dart"; +import 'package:expect/expect.dart'; +import 'package:ffi/ffi.dart'; + +import 'threading_utils.dart'; + +@pragma('vm:shared') +late Mutex mutexCondvar; +@pragma('vm:shared') +late ConditionVariable condVar; +@pragma('vm:shared') +bool greetingsReceived = false; + +int threadMain(Pointer data) { + final new_isolate = Isolate.create(debugName: "helper"); + Expect.isNotNull(new_isolate); + final SendPort sp = new_isolate.runSync(() { + Expect.isFalse(Isolate.current.isPinnedToCurrentThread); + Expect.isTrue(Isolate.pinToCurrentThread()); + Expect.isTrue(Isolate.current.isPinnedToCurrentThread); + + late RawReceivePort rp; + rp = RawReceivePort((e) { + print('running RawReceivePort handler $e'); + + Expect.isTrue(Isolate.current.isPinnedToCurrentThread); + + Expect.equals("greetings!", e); + mutexCondvar.runLocked(() { + greetingsReceived = true; + 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(greetingsReceived); + + print('=== running event loop for $new_isolate'); + new_isolate.runEventLoopSync(); + mutexCondvar.runLocked(() { + while (!greetingsReceived) { + condVar.wait(mutexCondvar); + } + }); + Expect.isTrue(greetingsReceived); + Expect.isTrue(new_isolate.isPinnedToCurrentThread); + + print('=== shutting down'); + new_isolate.shutdownSync(); + return 0; +} + +main(List args, SendPort? message) async { + if (Platform.isWindows) { + // pthread library loading doesn't work on Windows. + return; + } + + asyncStart(); + + mutexCondvar = Mutex(); + condVar = ConditionVariable(); + + final threadInfo = ThreadInfo(); + + Expect.equals(0, pthreadAttrInit(threadInfo.ptr_attr)); + threadInfo.ptr_data.cast()[0] = 0; + final callback = + NativeCallable)>.isolateGroupBound( + threadMain, + exceptionalReturn: -1, + ); + callback.keepIsolateAlive = false; + pthreadCreate( + threadInfo.ptr_tid, + threadInfo.ptr_attr, + callback.nativeFunction, + threadInfo.ptr_data.cast(), + ); + + threadInfo.join(); + + asyncEnd(); +}