[vm] Improve asynchronous unwinding through Stream methods

This CL adds @pragma('vm:awaiter-link') in various places in
Stream implementation to facilitate unwinding and expands
async unwinding logic with more information about
Stream internals.

At the same time be more conservative when checking if an
exception thrown from async method handled: failing to unwind
the stack fully creates situations when we incorrectly report
caught exceptions as uncaught, which frustrates users.

To distinguish stream subscriptions with and without error
handlers we add a state bit. Otherwise, it looks like all
subscriptions have error handlers because if no error
handler is installed we eagerly install error handler forwarding
the error to `Zone.handleUncaughtError`.

Fixes #53334
Fixes #54788
Fixes #47985

TEST=runtime/vm/dart/awaiter_stacks/stream_methods_test.dart,pkg/vm_service/test/pause_on_unhandled_async_exceptions6_test.dart,pkg/vm_service/test/pause_on_unhandled_async_exceptions7_test.dart

CoreLibraryReviewExempt: No behavioral change. Async changes reviewed by lrhn@
Cq-Include-Trybots: luci.dart.try:vm-aot-android-release-arm64c-try,vm-aot-android-release-arm_x64-try,vm-aot-dwarf-linux-product-x64-try,vm-aot-obfuscate-linux-release-x64-try,vm-aot-optimization-level-linux-release-x64-try
Change-Id: Ic51f926867092dd0adbe801b753f57c357c7ace2
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/322720
Reviewed-by: Alexander Markov <alexmarkov@google.com>
Reviewed-by: Lasse Nielsen <lrn@google.com>
Reviewed-by: Ben Konyi <bkonyi@google.com>
Commit-Queue: Slava Egorov <vegorov@google.com>

pkg/vm_service/test/common/service_test_common.dart

@@ -211,6 +211,62 @@ IsolateTest setBreakpointAtLineColumn(int line, int column) {
   };
 }
 
+extension FrameLocation on Frame {
+  Future<(String, int)> getLocation(
+    VmService service,
+    IsolateRef isolateRef,
+  ) async {
+    if (location?.tokenPos == null) {
+      return ('<unknown>', -1);
+    }
+
+    final script = (await service.getObject(
+      isolateRef.id!,
+      location!.script!.id!,
+    )) as Script;
+    return (
+      script.uri!,
+      script.getLineNumberFromTokenPos(location!.tokenPos!) ?? -1
+    );
+  }
+}
+
+Future<String> formatFrames(
+  VmService service,
+  IsolateRef isolateRef,
+  List<Frame> frames,
+) async {
+  final sb = StringBuffer();
+  for (Frame f in frames) {
+    sb.write(' $f');
+    if (f.function case final funcRef?) {
+      sb.write(' ');
+      sb.write(await qualifiedFunctionName(service, isolateRef, funcRef));
+    }
+    if (f.location != null) {
+      final (uri, lineNo) = await f.getLocation(service, isolateRef);
+      sb.write(' $uri:$lineNo');
+    }
+    sb.writeln();
+  }
+  return sb.toString();
+}
+
+Future<String> formatStack(
+  VmService service,
+  IsolateRef isolateRef,
+  Stack stack,
+) async {
+  final sb = StringBuffer();
+  sb.write('Full stack trace:\n');
+  sb.writeln(await formatFrames(service, isolateRef, stack.frames!));
+  if (stack.asyncCausalFrames case final asyncFrames?) {
+    sb.write('\nFull async stack trace:\n');
+    sb.writeln(await formatFrames(service, isolateRef, asyncFrames));
+  }
+  return sb.toString();
+}
+
 IsolateTest stoppedAtLine(int line) {
   return (VmService service, IsolateRef isolateRef) async {
     print('Checking we are at line $line');
@@ -227,33 +283,12 @@ IsolateTest stoppedAtLine(int line) {
     expect(frames.length, greaterThanOrEqualTo(1));
 
     final top = frames[0];
-    final Script script =
-        (await service.getObject(id, top.location!.script!.id!)) as Script;
-    final int actualLine =
-        script.getLineNumberFromTokenPos(top.location!.tokenPos!)!;
+    final (_, actualLine) = await top.getLocation(service, isolateRef);
     if (actualLine != line) {
       print('Actual: $actualLine Line: $line');
       final sb = StringBuffer();
       sb.write('Expected to be at line $line but actually at line $actualLine');
-      sb.write('\nFull stack trace:\n');
-      for (Frame f in frames) {
-        sb.write(
-          ' $f [${script.getLineNumberFromTokenPos(f.location!.tokenPos!)}]\n',
-        );
-      }
-      if (stack.asyncCausalFrames != null) {
-        final asyncFrames = stack.asyncCausalFrames!;
-        sb.write('\nFull async stack trace:\n');
-        for (Frame f in asyncFrames) {
-          sb.write(' $f');
-          if (f.location != null) {
-            sb.write(
-              ' [${script.getLineNumberFromTokenPos(f.location!.tokenPos!)}]',
-            );
-          }
-          sb.writeln();
-        }
-      }
+      sb.writeln(await formatStack(service, isolateRef, stack));
       throw sb.toString();
     } else {
       print('Program is stopped at line: $line');
@@ -708,22 +743,35 @@ IsolateTest stoppedInFunction(String functionName) {
         'Expected to be in function $functionName but '
         'actually in function $name',
       );
-      sb.writeln('Full stack trace:');
-      for (final frame in frames) {
-        final func = await service.getObject(
-          isolateId,
-          frame.function!.id!,
-        ) as Func;
-        final ownerName = func.owner.name!;
-        sb.write(' $frame [${func.name}] [$ownerName]\n');
-      }
+      sb.writeln(await formatStack(service, isolateRef, stack));
+
       throw sb.toString();
     } else {
       print('Program is stopped in function: $functionName');
     }
   };
 }
 
+Future<String> qualifiedFunctionName(
+  VmService service,
+  IsolateRef isolate,
+  FuncRef func,
+) async {
+  final funcName = func.name ?? '<unknown>';
+  switch (func.owner) {
+    case final FuncRef parentFuncRef:
+      final parentFuncName =
+          await qualifiedFunctionName(service, isolate, parentFuncRef);
+      return '$parentFuncName.$funcName';
+
+    case final ClassRef parentClass:
+      return '${parentClass.name!}.$funcName';
+
+    case _:
+      return funcName;
+  }
+}
+
 Future<void> expectFrame(
   VmService service,
   IsolateRef isolate,
@@ -734,7 +782,10 @@ Future<void> expectFrame(
 }) async {
   expect(frame.kind, equals(kind));
   if (functionName != null) {
-    expect(frame.function?.name, equals(functionName));
+    expect(
+      await qualifiedFunctionName(service, isolate, frame.function!),
+      equals(functionName),
+    );
   }
   if (line != null) {
     expect(frame.location, isNotNull);
@@ -749,3 +800,68 @@ Future<void> expectFrame(
     );
   }
 }
+
+Future<String> getCurrentExceptionAsString(
+  VmService service,
+  IsolateRef isolateRef,
+) async {
+  final isolate = await service.getIsolate(isolateRef.id!);
+  final event = isolate.pauseEvent!;
+  final exception = await service.getObject(
+    isolateRef.id!,
+    event.exception!.id!,
+  ) as Instance;
+  return exception.valueAsString!;
+}
+
+typedef ExpectedFrame = ({String? functionName, int? line});
+const ExpectedFrame asyncGap = (functionName: null, line: null);
+
+IsolateTest resumePastUnhandledException(String exceptionAsString) {
+  return (service, isolateRef) async {
+    do {
+      await resumeIsolate(service, isolateRef);
+      await hasStoppedWithUnhandledException(service, isolateRef);
+    } while (await getCurrentExceptionAsString(service, isolateRef) ==
+        exceptionAsString);
+  };
+}
+
+IsolateTest expectUnhandledExceptionWithFrames({
+  List<ExpectedFrame>? expectedFrames,
+  String? exceptionAsString,
+}) {
+  return (VmService service, IsolateRef isolateRef) async {
+    await hasStoppedWithUnhandledException(service, isolateRef);
+    if (exceptionAsString != null) {
+      expect(
+        await getCurrentExceptionAsString(service, isolateRef),
+        equals(exceptionAsString),
+      );
+    }
+
+    if (expectedFrames == null) {
+      return;
+    }
+
+    final stack = await service.getStack(isolateRef.id!);
+
+    final frames = stack.asyncCausalFrames!;
+    var currentKind = 'Regular';
+    for (var i = 0; i < expectedFrames.length; i++) {
+      final expected = expectedFrames[i];
+      final got = frames[i];
+      await expectFrame(
+        service,
+        isolateRef,
+        got,
+        kind: expected == asyncGap ? 'AsyncSuspensionMarker' : currentKind,
+        functionName: expected.functionName,
+        line: expected.line,
+      );
+      if (expected == asyncGap) {
+        currentKind = 'AsyncCausal';
+      }
+    }
+  };
+}

pkg/vm_service/test/common/test_helper.dart

@@ -337,8 +337,17 @@ class _ServiceTesterRunner {
       timeout: Timeout.none,
     );
 
-    tearDown(() {
+    tearDown(() async {
       print('All service tests completed successfully.');
+      try {
+        await vm.dispose();
+      } catch (e, st) {
+        print('''
+Ignoring exception during vm-service connection shutdown:
+$e
+$st
+''');
+      }
       process.requestExit();
     });