[vm/shared] Add timeout to ConditionVariable.wait.

Update isolate_group_shared_callback_test to use that and also use flag to print exceptions.

TEST=ci
CoreLibraryReviewExempt: only internal library is affected
Change-Id: Ia53ca7a6dc847b551229ccf00bbbfe302a260fcb
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/427680
Reviewed-by: Slava Egorov <[email protected]>
Commit-Queue: Alexander Aprelev <[email protected]>

Author: aam

runtime/bin/ffi_test/ffi_test_functions.cc

@@ -1342,7 +1342,7 @@ static void* Start(void* data_in) {
   Data* data = reinterpret_cast<Data*>(data_in);
   void (*fn)(int64_t, int32_t) = data->fn;
   int64_t a = data->a;
-  int64_t b = data->b;
+  int32_t b = data->b;
   delete data;
   fn(a, b);
   return nullptr;

runtime/lib/concurrent.cc

@@ -67,7 +67,8 @@ DEFINE_FFI_NATIVE_ENTRY(ConditionVariable_Initialize,
 DEFINE_FFI_NATIVE_ENTRY(ConditionVariable_Wait,
                         void,
                         (Dart_Handle condvar_handle,
-                         Dart_Handle mutex_handle)) {
+                         Dart_Handle mutex_handle,
+                         intptr_t timeout)) {
   Mutex* mutex;
   Dart_Handle result_mutex = Dart_GetNativeInstanceField(
       mutex_handle, kCondVarNativeField, reinterpret_cast<intptr_t*>(&mutex));
@@ -81,7 +82,7 @@ DEFINE_FFI_NATIVE_ENTRY(ConditionVariable_Wait,
   if (Dart_IsError(result_condvar)) {
     Dart_PropagateError(result_condvar);
   }
-  condvar->Wait(mutex);
+  condvar->Wait(mutex, timeout);
 }
 
 DEFINE_FFI_NATIVE_ENTRY(ConditionVariable_Notify,

runtime/vm/bootstrap_natives.h

@@ -374,7 +374,7 @@ namespace dart {
   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(ConditionVariable_Wait, void, (Dart_Handle, Dart_Handle, intptr_t))        \
   V(FinalizerEntry_SetExternalSize, void, (Dart_Handle, intptr_t))             \
   V(IsolateGroup_runSync, Dart_Handle, (Dart_Handle))                          \
   V(Mutex_Initialize, void, (Dart_Handle))                                     \

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

@@ -3,7 +3,7 @@
 // BSD-style license that can be found in the LICENSE file.
 
 import "dart:_internal" show patch;
-import "dart:ffi" show Handle, Void, Native;
+import "dart:ffi" show IntPtr, Handle, Native, Void;
 import "dart:nativewrappers" show NativeFieldWrapperClass1;
 
 @patch
@@ -47,8 +47,17 @@ base class _ConditionVariableImpl extends NativeFieldWrapperClass1
   @Native<Void Function(Handle)>(symbol: "ConditionVariable_Initialize")
   external void _initialize();
 
-  @Native<Void Function(Handle, Handle)>(symbol: "ConditionVariable_Wait")
-  external void wait(Mutex mutex);
+  @Native<Void Function(Handle, Handle, IntPtr)>(
+    symbol: "ConditionVariable_Wait",
+  )
+  external void _wait(Mutex mutex, int timeout);
+
+  void wait(Mutex mutex, [int timeout = 0]) {
+    if (timeout < 0) {
+      throw ArgumentError.value(timeout, "timeout", "must be positive or zero");
+    }
+    _wait(mutex, timeout);
+  }
 
   @Native<Void Function(Handle)>(symbol: "ConditionVariable_Notify")
   external void notify();

sdk/lib/concurrent/concurrent.dart

@@ -46,7 +46,13 @@ abstract interface class ConditionVariable {
   /// `mutex` must be a [Mutex] object exclusively held by the current thread.
   /// It will be released and the thread will block until another thread
   /// calls [notify].
-  external void wait(Mutex mutex);
+  ///
+  /// If `timeout` is provided, it must be positive or zero.
+  /// Default zero value indicates that the method will wait to be notified
+  /// infinitely, without timeout.
+  /// If it is greater than zero, then after this many milliseconds the method
+  /// will return even if the variable was not notified.
+  external void wait(Mutex mutex, [int timeout = 0]);
 
   /// Wake up at least one thread waiting on this condition variable.
   external void notify();

tests/ffi/isolate_group_shared_callback_test.dart

@@ -4,19 +4,20 @@
 
 // Dart test program for testing dart:ffi async callbacks.
 //
-// VMOptions=--experimental-shared-data
-// VMOptions=--experimental-shared-data --use-slow-path
-// VMOptions=--experimental-shared-data --use-slow-path --stacktrace-every=100
-// VMOptions=--experimental-shared-data --dwarf_stack_traces --no-retain_function_objects --no-retain_code_objects
-// VMOptions=--experimental-shared-data --test_il_serialization
-// VMOptions=--experimental-shared-data --profiler --profile_vm=true
-// VMOptions=--experimental-shared-data --profiler --profile_vm=false
+// VMOptions=--experimental-shared-data --print-stacktrace-at-throw
+// VMOptions=--experimental-shared-data --print-stacktrace-at-throw --use-slow-path
+// VMOptions=--experimental-shared-data --print-stacktrace-at-throw --use-slow-path --stacktrace-every=100
+// VMOptions=--experimental-shared-data --print-stacktrace-at-throw --dwarf_stack_traces --no-retain_function_objects --no-retain_code_objects
+// VMOptions=--experimental-shared-data --print-stacktrace-at-throw --test_il_serialization
+// VMOptions=--experimental-shared-data --print-stacktrace-at-throw --profiler --profile_vm=true
+// VMOptions=--experimental-shared-data --print-stacktrace-at-throw --profiler --profile_vm=false
 // SharedObjects=ffi_test_functions
 
 import 'dart:async';
 import 'dart:concurrent';
 import 'dart:ffi';
 import 'dart:isolate';
+import 'package:dart_internal/isolate_group.dart' show IsolateGroup;
 
 import 'dart:io';
 
@@ -45,14 +46,6 @@ class NativeLibrary {
   late final FnSleepType sleep;
 
   NativeLibrary() {
-    callFunctionOnSameThread = ffiTestFunctions
-        .lookupFunction<FnRunnerNativeType, FnRunnerType>(
-          "CallFunctionOnSameThread",
-        );
-    callFunctionOnNewThreadBlocking = ffiTestFunctions
-        .lookupFunction<FnRunnerNativeType, FnRunnerType>(
-          "CallFunctionOnNewThreadBlocking",
-        );
     callFunctionOnNewThreadNonBlocking = ffiTestFunctions
         .lookupFunction<FnRunnerNativeType, FnRunnerType>(
           "CallFunctionOnNewThreadNonBlocking",
@@ -92,7 +85,6 @@ void simpleFunction(int a, int b) {
 Future<void> testNativeCallableHelloWorld() async {
   mutexCondvar = Mutex();
   conditionVariable = ConditionVariable();
-  // final callback = NativeCallable<CallbackNativeType>.isolateGroupShared(simpleFunction);
   final callback = NativeCallable<CallbackNativeType>.isolateGroupShared(
     simpleFunction,
   );
@@ -103,7 +95,8 @@ Future<void> testNativeCallableHelloWorld() async {
 
   mutexCondvar.runLocked(() {
     while (!resultIsReady) {
-      conditionVariable.wait(mutexCondvar);
+      conditionVariable.wait(mutexCondvar, 10 * sleepForMs);
+      Expect.isTrue(resultIsReady);
     }
   });
 
@@ -113,16 +106,38 @@ Future<void> testNativeCallableHelloWorld() async {
   lib.callFunctionOnNewThreadNonBlocking(1001, callback.nativeFunction);
   mutexCondvar.runLocked(() {
     while (!resultIsReady) {
-      conditionVariable.wait(mutexCondvar);
+      conditionVariable.wait(mutexCondvar, 10 * sleepForMs);
+      Expect.isTrue(resultIsReady);
     }
   });
   Expect.equals(42 + (1001 * 123) * 2, result);
 }
 
+void simpleFunctionThatThrows(int a, int b) {
+  // Complete without notifying mutexCondvar
+  throw 'hello, world';
+}
+
+Future<void> testNativeCallableThrows() async {
+  mutexCondvar = Mutex();
+  conditionVariable = ConditionVariable();
+  final callback = NativeCallable<CallbackNativeType>.isolateGroupShared(
+    simpleFunctionThatThrows,
+  );
+
+  result = 42;
+  resultIsReady = false;
+  lib.callFunctionOnNewThreadNonBlocking(1001, callback.nativeFunction);
+
+  mutexCondvar.runLocked(() {
+    conditionVariable.wait(mutexCondvar, 10 * sleepForMs);
+    Expect.isFalse(resultIsReady);
+  });
+}
+
 Future<void> testNativeCallableHelloWorldClosure() async {
   mutexCondvar = Mutex();
   conditionVariable = ConditionVariable();
-  // final callback = NativeCallable<CallbackNativeType>.isolateGroupShared(simpleFunction);
   final callback = NativeCallable<CallbackNativeType>.isolateGroupShared((
     int a,
     int b,
@@ -213,6 +228,7 @@ main(args, message) async {
   lib = NativeLibrary();
   // Simple tests.
   await testNativeCallableHelloWorld();
+  await testNativeCallableThrows();
   await testNativeCallableHelloWorldClosure();
   testNativeCallableSync();
   testNativeCallableSyncThrows();