[vm] Test active mutator stealing works with dart:concurrent locks.

 - Add Monitor.notifyAll.
 - Refactor Mutex.runLocked to unlock if interrupted by unwind error.
 - Fix skipped DLRT_ExitHandleScope during an unwind error.

TEST=vm/dart/isolates/many_isolates_blocked_at_monitor_test
Bug: #55991
CoreLibraryReviewExempt: VM-only experimental library
Change-Id: I7a37544436e22158e45ee44567600fcf7ff7d7b6
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/409561
Reviewed-by: Alexander Aprelev <[email protected]>
Commit-Queue: Ryan Macnak <[email protected]>

Author: rmacnak-google

runtime/lib/concurrent.cc

@@ -25,24 +25,23 @@ DEFINE_FFI_NATIVE_ENTRY(Mutex_Initialize, void, (Dart_Handle mutex_handle)) {
   Dart_NewFinalizableHandle(mutex_handle, mutex, sizeof(Mutex), DeleteMutex);
 };
 
-DEFINE_FFI_NATIVE_ENTRY(Mutex_Lock, void, (Dart_Handle mutex_handle)) {
+DEFINE_FFI_NATIVE_ENTRY(Mutex_RunLocked,
+                        Dart_Handle,
+                        (Dart_Handle mutex_handle,
+                         Dart_Handle closure_handle)) {
   Mutex* mutex;
   Dart_Handle result = Dart_GetNativeInstanceField(
       mutex_handle, kMutexNativeField, reinterpret_cast<intptr_t*>(&mutex));
   if (Dart_IsError(result)) {
     Dart_PropagateError(result);
   }
   mutex->Lock();
-}
-
-DEFINE_FFI_NATIVE_ENTRY(Mutex_Unlock, void, (Dart_Handle mutex_handle)) {
-  Mutex* mutex;
-  Dart_Handle result = Dart_GetNativeInstanceField(
-      mutex_handle, kMutexNativeField, reinterpret_cast<intptr_t*>(&mutex));
+  result = Dart_InvokeClosure(closure_handle, 0, nullptr);
+  mutex->Unlock();
   if (Dart_IsError(result)) {
     Dart_PropagateError(result);
   }
-  mutex->Unlock();
+  return result;
 }
 
 static void DeleteConditionVariable(void* isolate_data, void* condvar_pointer) {
@@ -98,4 +97,17 @@ DEFINE_FFI_NATIVE_ENTRY(ConditionVariable_Notify,
   condvar->Notify();
 }
 
+DEFINE_FFI_NATIVE_ENTRY(ConditionVariable_NotifyAll,
+                        void,
+                        (Dart_Handle condvar_handle)) {
+  ConditionVariable* condvar;
+  Dart_Handle result_condvar =
+      Dart_GetNativeInstanceField(condvar_handle, kCondVarNativeField,
+                                  reinterpret_cast<intptr_t*>(&condvar));
+  if (Dart_IsError(result_condvar)) {
+    Dart_PropagateError(result_condvar);
+  }
+  condvar->NotifyAll();
+}
+
 }  // namespace dart

runtime/tests/vm/dart/isolates/many_isolates_blocked_at_monitor_test.dart

@@ -0,0 +1,47 @@
+// Copyright (c) 2025, 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.
+
+// VMOptions=--experimental-shared-data
+
+import "dart:concurrent";
+import "dart:isolate";
+import "dart:io";
+
+@pragma("vm:shared")
+int pending = 0;
+@pragma("vm:shared")
+late Mutex mutex;
+@pragma("vm:shared")
+late ConditionVariable condition;
+
+waitForAllToCheckIn() {
+  mutex.runLocked(() {
+    pending--;
+    if (pending == 0) {
+      print("notifyAll $pending");
+      condition.notifyAll();
+    } else {
+      while (pending > 0) {
+        print("wait $pending");
+        condition.wait(mutex);
+        print("notified $pending");
+      }
+    }
+    condition.notifyAll();
+  });
+}
+
+child(_) {
+  waitForAllToCheckIn();
+}
+
+main() async {
+  mutex = new Mutex();
+  condition = new ConditionVariable();
+  pending = 21;
+  for (var i = 0; i < 20; i++) {
+    Isolate.spawn(child, null);
+  }
+  waitForAllToCheckIn();
+}

runtime/vm/bootstrap_natives.h

@@ -368,11 +368,11 @@ namespace dart {
 #define BOOTSTRAP_FFI_NATIVE_LIST(V)                                           \
   V(ConditionVariable_Initialize, void, (Dart_Handle))                         \
   V(ConditionVariable_Notify, void, (Dart_Handle))                             \
+  V(ConditionVariable_NotifyAll, void, (Dart_Handle))                          \
   V(ConditionVariable_Wait, void, (Dart_Handle, Dart_Handle))                  \
   V(FinalizerEntry_SetExternalSize, void, (Dart_Handle, intptr_t))             \
   V(Mutex_Initialize, void, (Dart_Handle))                                     \
-  V(Mutex_Lock, void, (Dart_Handle))                                           \
-  V(Mutex_Unlock, void, (Dart_Handle))                                         \
+  V(Mutex_RunLocked, Dart_Handle, (Dart_Handle, Dart_Handle))                  \
   V(Pointer_asTypedListFinalizerAllocateData, void*, ())                       \
   V(Pointer_asTypedListFinalizerCallbackPointer, void*, ())
 

runtime/vm/dart_entry.cc

@@ -52,9 +52,18 @@ class DartEntryScope : public TransitionToGenerated {
     saved_safestack_limit_ = OSThread::GetCurrentSafestackPointer();
     thread->set_saved_safestack_limit(saved_safestack_limit_);
 #endif
+
+    saved_api_scope_ = thread->api_top_scope();
   }
 
   ~DartEntryScope() {
+    // Propagating an Error that is not an UnhandledException, such as an
+    // UnwindError, will bypass the DLRT_ExitHandleScope in an FFI callout
+    // trampoline.
+    while (UNLIKELY(thread()->api_top_scope() != saved_api_scope_)) {
+      thread()->ExitApiScope();
+    }
+
 #if defined(USING_SAFE_STACK)
     thread()->set_saved_safestack_limit(saved_safestack_limit_);
 #endif
@@ -68,6 +77,7 @@ class DartEntryScope : public TransitionToGenerated {
 #if defined(USING_SAFE_STACK)
   uword saved_safestack_limit_ = 0;
 #endif
+  ApiLocalScope* saved_api_scope_;
 };
 
 // Note: The invocation stub follows the C ABI, so we cannot pass C++ struct

sdk/lib/_internal/vm/lib/concurrent_patch.dart

@@ -23,20 +23,11 @@ base class _MutexImpl extends NativeFieldWrapperClass1 implements Mutex {
   external void _initialize();
 
   @patch
-  @Native<Void Function(Handle)>(symbol: "Mutex_Lock")
-  external void _lock();
-
-  @patch
-  @Native<Void Function(Handle)>(symbol: "Mutex_Unlock")
-  external void _unlock();
+  @Native<Handle Function(Handle, Handle)>(symbol: "Mutex_RunLocked")
+  external Object _runLocked(Object action);
 
   R runLocked<R>(R Function() action) {
-    _lock();
-    try {
-      return action();
-    } finally {
-      _unlock();
-    }
+    return _runLocked(action) as R;
   }
 }
 
@@ -64,4 +55,8 @@ base class _ConditionVariableImpl extends NativeFieldWrapperClass1
   @patch
   @Native<Void Function(Handle)>(symbol: "ConditionVariable_Notify")
   external void notify();
+
+  @patch
+  @Native<Void Function(Handle)>(symbol: "ConditionVariable_NotifyAll")
+  external void notifyAll();
 }

sdk/lib/concurrent/concurrent.dart

@@ -19,21 +19,7 @@ abstract interface class Mutex {
 
   external factory Mutex._();
 
-  /// Acquire exclusive ownership of this mutex.
-  ///
-  /// If this mutex is already acquired then an attempt to acquire it
-  /// blocks the current thread until the mutex is released by the
-  /// current owner.
-  ///
-  /// **Warning**: attempting to hold a mutex across asynchronous suspension
-  /// points will lead to undefined behavior and potentially crashes.
-  external void _lock();
-
-  /// Release exclusive ownership of this mutex.
-  ///
-  /// It is an error to release ownership of the mutex if it was not
-  /// previously acquired.
-  external void _unlock();
+  external Object _runLocked(Object action);
 
   /// Run the given synchronous `action` under a mutex.
   ///
@@ -64,4 +50,7 @@ abstract interface class ConditionVariable {
 
   /// Wake up at least one thread waiting on this condition variable.
   external void notify();
+
+  /// Wake up all threads waiting on this condition variable.
+  external void notifyAll();
 }