[vm] Rework awaiter stack unwinding.

The main contribution of this CL is unification of disparate
handling of various functions like `Future.timeout`,
`Future.wait`, `_SuspendState.createAsyncCallbacks` and
`_SuspendState._createAsyncStarCallback` into a single
`@pragma('vm:awaiter-link')` which allows Dart developers
to specify where awaiter unwinder should look for the next
awaiter.

For example this allows unwinding to succeed for the code like this:

    Future<int> outer(Future<int> inner) {
      @pragma('vm:awaiter-link')
      final completer = Completer<int>();

      inner.then((v) => completer.complete(v));

      return completer.future;
   }

This refactoring also ensures that we preserve information
(including Function & Code objects) required for awaiter
unwinding across all modes (JIT, AOT and AOT with DWARF stack
traces). This guarantees users will get the same information
no matter which mode they are running in. Previously
we have been disabling awaiter_stacks tests in some AOT
modes - which led to regressions in the quality of produced
stacks.

This CL also cleans up relationship between debugger and awaiter
stack returned by StackTrace.current - which makes stack trace
displayed by debugger (used for stepping out and determinining
whether exception is caught or not) and `StackTrace.current`
consistent.

Finally we make one user visible change to the stack trace:
awaiter stack will no always include intermediate listeners
created through `Future.then`. Previously we would sometimes
include these listeners at the tail of the stack trace,
which was inconsistent.

Ultimately this means that code like this:

    Future<int> inner() async {
      await null;  // asynchronous gap
      print(StackTrace.current); // (*)
      return 0;
    }

    Future<int> outer() async {
      int process(int v) {
        return v + 1;
      }

      return await inner().then(process);
    }

    void main() async {
      await outer();
    }

Produces stack trace like this:

    inner
    <asynchronous suspension>
    outer.process
    <asynchronous suspension>
    outer
    <asynchronous suspension>
    main
    <asynchronous suspension>

And when stepping out of `inner` execution will stop at `outer.process`
first and the next step out will bring execution to `outer` next.

Fixes #52797
Fixes #52203
Issue #47985

TEST=ci

Bug: b/279929839
CoreLibraryReviewExempt: CL just adds @pragma to facilitate unwinding
Cq-Include-Trybots: luci.dart.try:vm-aot-linux-product-x64-try,vm-aot-linux-debug-x64-try,vm-aot-linux-release-x64-try,vm-aot-obfuscate-linux-release-x64-try,vm-aot-dwarf-linux-product-x64-try
Change-Id: If377d5329d6a11c86effb9369dc603a7ae616fe7
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/311680
Reviewed-by: Alexander Markov <alexmarkov@google.com>
Commit-Queue: Slava Egorov <vegorov@google.com>

pkg/native_stack_traces/lib/src/dwarf.dart

@@ -816,7 +816,11 @@ class DebugInformationEntry {
       DartCallInfo(
           function: unit.nameOfOrigin(abstractOrigin ?? -1),
           inlined: inlined,
-          internal: isArtificial,
+          // Don't hide artificial (invisible) methods which appear as
+          // Future listeners. This is because tear-offs of static methods
+          // are marked as invisible (even if the method itself is visible)
+          // and we want these to appear in stack traces.
+          internal: isArtificial && address != lowPC,
           filename: filename,
           line: line,
           column: column)

pkg/vm_service/test/get_stack_test.dart

@@ -89,12 +89,11 @@ final tests = <IsolateTest>[
   (VmService service, IsolateRef isolateRef) async {
     final result = await service.getStack(isolateRef.id!);
 
-    expect(result.frames, hasLength(6));
+    expect(result.frames, hasLength(5));
     expect(result.asyncCausalFrames, hasLength(26));
 
     expectFrames(result.frames!, [
       [equals('Regular'), endsWith(' func10')],
-      [equals('Regular'), endsWith(' _RootZone.runUnary')],
       [equals('Regular'), anything], // Internal mech. ..
       [equals('Regular'), anything],
       [equals('Regular'), anything],

runtime/docs/awaiter_stack_traces.md

@@ -0,0 +1,127 @@
+# Awaiter Stack Traces
+
+One of the common challenges associated with debugging asynchronous code is that stack traces do not reference the code which led to the exception. The context is lost when execution cross asynchronous gap.
+
+Consider the following code:
+
+```dart
+Future<int> inner() async {
+  await null;  // asynchronous gap
+  print(StackTrace.current); // (*)
+  return 0;
+}
+
+Future<int> outer() async {
+  int process(int v) {
+    return v + 1;
+  }
+
+  return await inner().then(process);
+}
+
+void main() async {
+  await outer();
+}
+```
+
+Producing synchronous stack trace at the line marked `(*)` will yield the following:
+
+```
+#0 inner
+#1 _SuspendState._createAsyncCallbacks.thenCallback
+#2 _RootZone.runUnary
+#3 _SuspendState._awaitNotFuture.run
+#4 _microtaskLoop
+#5 _startMicrotaskLoop
+#6 _startMicrotaskLoop
+#7 _runPendingImmediateCallback
+#8 _RawReceivePort._handleMessage
+```
+
+Only a single frame corresponds to user code (`#0  inner`) and the rest are `dart:async` internals. Nothing in this stack trace mentions `outer` or `main`, which called `inner`. This makes diagnosing issues based on a stack trace much harder.
+
+To address this problem runtime system augments synchronous portion of the stack trace with an _awaiter stack trace_. Each awaiter frame represents a closure or a suspended `async` function which will be invoked when the currently running asynchronous computation completes.
+
+This support allows runtime system to produce the following output for the example given above:
+
+```
+#0  inner
+<asynchronous suspension>
+#1  outer.process
+<asynchronous suspension>
+#2  outer
+<asynchronous suspension>
+#3  main
+<asynchronous suspension>
+```
+
+## Algorithm
+
+To recover awaiter stack trace runtime follows through a chain of `_Future`, `_StreamController` and `SuspendState` objects. The following diagram illustrates the path it takes to produce asynchronous stack trace in our initial example:
+
+![Heap structure used for unwinding](awaiter_stack_unwinding.png)
+
+Each awaiter frame is a pair of `(closure, nextFrame)`:
+
+* `closure` is a listener which will be invoked when the future this frame is waiting on completes.
+  * This might be one of the callbacks associated with [suspendable functions](async.md) internals, e.g. `_SuspendState.thenCallback` which resumes execution after the `await`.
+* `next` is an object representing the next awaiter frame, which is waiting for the completion of this awaiter frame.
+
+Unwinding rules can be summarised as follows:
+
+* If at any point `closure` has a captured variable marked with `@pragma('vm:awaiter-link')` variable then the value of that variable will be used as `nextFrame`.
+* If `nextFrame` is a `_SuspendState` then `_SuspendState.function_data` gives us `_FutureImpl` or `_AsyncStarStreamController` to look at.
+* If `nextFrame` is `_FutureImpl` then we can take the first `_FutureListener` in `listeners` and then the next frame is `(listener.callback, listener.result)`.
+* If `nextFrame` is `_AsyncStarStreamController` then we get  `asyncStarStreamController.controller.subscription._onData`, which should give us an instance of `_StreamIterator`, which inside contains a `_FutureImpl` on which `await for` is waiting.
+
+Awaiter unwinding is implemented in by [`dart::StackTraceUtils::CollectFrames`] in [`runtime/vm/stack_trace.cc`].
+
+### `@pragma('vm:awaiter-link')`
+
+Dart code which does not use `async`/`async*` functions and instead uses callbacks and lower-level primitives can integrate with awaiter frame unwinding by annotating variables which link to the next awaiter frame with `@pragma('vm:awaiter-link')`.
+
+Consider the following variation of the example:
+
+```dart
+Future<int> outer() {
+  final completer = Completer<int>();
+
+  int process(int v) {
+    completer.complete(v);
+  }
+
+  inner().then(v);
+
+  return completer.future;
+}
+```
+
+Running this would produce the following stack trace:
+
+```
+#0 inner
+<asynchronous suspension>
+#1 outer.process
+<asynchronous suspension>
+```
+
+Runtime is unable to unwind the awaiter stack past `process`. However if `completer` is annotated with `@pragma('vm:awaiter-link')` then unwinder will know where to continue:
+
+```dart
+Future<int> outer() {
+  @pragma('vm:awaiter-link')
+  final completer = Completer<int>();
+  // ...
+}
+```
+
+```
+#0      inner
+<asynchronous suspension>
+#1      outer.process
+<asynchronous suspension>
+#2      main
+<asynchronous suspension>
+```
+
+`vm:awaiter-link` can be used in `dart:async` internals to avoid hardcoding recognition of specific methods into the runtime system, see for example `_SuspendState.thenCallback` or `Future.timeout` implementations.

runtime/docs/pragmas.md

@@ -17,6 +17,7 @@ These pragmas are part of the VM's API and are safe for use in external code.
 | `vm:platform-const` | Marks a static getter or a static field with an initializer where the getter body or field initializer evaluates to a constant value if the target operating system is known. |
 | `weak-tearoff-reference` | [Declaring a static weak reference intrinsic method.](compiler/pragmas_recognized_by_compiler.md#declaring-a-static-weak-reference-intrinsic-method) |
 | `vm:isolate-unsendable` | Marks a class, instances of which won't be allowed to be passed through ports or sent between isolates. |
+| `vm:awaiter-link` | [Specifying variable to follow for awaiter stack unwinding](awaiter_stack_traces.md) |
 
 ## Unsafe pragmas for general use
 

runtime/lib/stacktrace.cc

@@ -41,11 +41,14 @@ static StackTracePtr CurrentStackTrace(Thread* thread,
 
   const auto& code_array = GrowableObjectArray::ZoneHandle(
       zone, GrowableObjectArray::New(kDefaultStackAllocation));
-  GrowableArray<uword> pc_offset_array;
+  GrowableArray<uword> pc_offset_array(kDefaultStackAllocation);
 
   // Collect the frames.
-  StackTraceUtils::CollectFrames(thread, code_array, &pc_offset_array,
-                                 skip_frames);
+  StackTraceUtils::CollectFrames(thread, skip_frames,
+                                 [&](const StackTraceUtils::Frame& frame) {
+                                   code_array.Add(frame.code);
+                                   pc_offset_array.Add(frame.pc_offset);
+                                 });
 
   return CreateStackTraceObject(zone, code_array, pc_offset_array);
 }

runtime/observatory/tests/service/async_step_out_test.dart

@@ -14,13 +14,15 @@ import 'test_helper.dart';
 //
 // dart runtime/observatory/tests/service/update_line_numbers.dart <test.dart>
 //
-const int LINE_A = 27;
-const int LINE_B = 28;
-const int LINE_C = 29;
-const int LINE_0 = 38;
-const int LINE_D = 39;
-const int LINE_E = 40;
-const int LINE_F = 41;
+const int LINE_A = 29;
+const int LINE_B = 30;
+const int LINE_C = 31;
+const int LINE_G = 36;
+const int LINE_0 = 40;
+const int LINE_D = 41;
+const int LINE_E = 42;
+const int LINE_F = 43;
+const int LINE_H = 44;
 // AUTOGENERATED END
 
 Future<int> helper() async {
@@ -31,14 +33,15 @@ Future<int> helper() async {
 }
 
 Future<void> testMain() async {
-  int process(int value) {
+  int process(int value) /* LINE_G */ {
     return value + 1;
   }
 
   debugger(); // LINE_0.
   print('line d'); // LINE_D.
   await helper().then(process); // LINE_E.
-  print('line f'); // LINE_F.
+  final v = process(10); // LINE_F.
+  print('line h'); // LINE_H.
 }
 
 var tests = <IsolateTest>[
@@ -66,8 +69,16 @@ var tests = <IsolateTest>[
   stoppedAtLine(LINE_C),
   stepOut, // out of helper to awaiter testMain.
 
+  hasStoppedAtBreakpoint,
+  stoppedAtLine(LINE_G),
+  stepOut, // out of helper to awaiter testMain.
+
   hasStoppedAtBreakpoint,
   stoppedAtLine(LINE_F),
+  stepOver,
+
+  hasStoppedAtBreakpoint,
+  stoppedAtLine(LINE_H),
 ];
 
 main(args) => runIsolateTests(args, tests,

runtime/observatory/tests/service/get_stack_limit_rpc_test.dart

@@ -53,10 +53,18 @@ var tests = <IsolateTest>[
     expect(asyncFrames.length, greaterThan(frames.length));
     expect(stack['truncated'], false);
     verifyStack(frames, [
-      'bar', 'foo', 'bar', 'foo',
-      'bar', 'foo', 'bar', 'foo',
-      'bar', 'foo', 'bar', 'foo',
-      '_RootZone.runUnary', // Internal async. mech. ..
+      'bar',
+      'foo',
+      'bar',
+      'foo',
+      'bar',
+      'foo',
+      'bar',
+      'foo',
+      'bar',
+      'foo',
+      'bar',
+      'foo'
     ]);
 
     final fullStackLength = frames.length;
@@ -71,10 +79,18 @@ var tests = <IsolateTest>[
     expect(asyncFrames.length, fullStackLength + 1);
     expect(stack['truncated'], true);
     verifyStack(frames, [
-      'bar', 'foo', 'bar', 'foo',
-      'bar', 'foo', 'bar', 'foo',
-      'bar', 'foo', 'bar', 'foo',
-      '_RootZone.runUnary', // Internal async. mech. ..
+      'bar',
+      'foo',
+      'bar',
+      'foo',
+      'bar',
+      'foo',
+      'bar',
+      'foo',
+      'bar',
+      'foo',
+      'bar',
+      'foo'
     ]);
 
     // Try a limit < actual stack depth and expect to get a stack of depth

runtime/observatory_2/tests/service_2/async_step_out_test.dart

@@ -4,6 +4,8 @@
 //
 // VMOptions=--async-debugger --verbose-debug
 
+// @dart=2.9
+
 import 'dart:developer';
 import 'service_test_common.dart';
 import 'test_helper.dart';
@@ -14,30 +16,34 @@ import 'test_helper.dart';
 //
 // dart runtime/observatory/tests/service/update_line_numbers.dart <test.dart>
 //
-const int LINE_A = 27;
-const int LINE_B = 28;
-const int LINE_C = 29;
-const int LINE_0 = 37;
-const int LINE_D = 38;
-const int LINE_E = 39;
-const int LINE_F = 40;
+const int LINE_A = 31;
+const int LINE_B = 32;
+const int LINE_C = 33;
+const int LINE_G = 38;
+const int LINE_0 = 42;
+const int LINE_D = 43;
+const int LINE_E = 44;
+const int LINE_F = 45;
+const int LINE_H = 46;
 // AUTOGENERATED END
 
-helper() async {
+Future<int> helper() async {
   await null; // LINE_A.
   print('line b'); // LINE_B.
   print('line c'); // LINE_C.
+  return 0;
 }
 
-testMain() async {
-  int process(int value) {
+Future<void> testMain() async {
+  int process(int value) /* LINE_G */ {
     return value + 1;
   }
 
   debugger(); // LINE_0.
   print('line d'); // LINE_D.
   await helper().then(process); // LINE_E.
-  print('line f'); // LINE_F.
+  final v = process(10); // LINE_F.
+  print('line h'); // LINE_H.
 }
 
 var tests = <IsolateTest>[
@@ -65,8 +71,16 @@ var tests = <IsolateTest>[
   stoppedAtLine(LINE_C),
   stepOut, // out of helper to awaiter testMain.
 
+  hasStoppedAtBreakpoint,
+  stoppedAtLine(LINE_G),
+  stepOut, // out of helper to awaiter testMain.
+
   hasStoppedAtBreakpoint,
   stoppedAtLine(LINE_F),
+  stepOver,
+
+  hasStoppedAtBreakpoint,
+  stoppedAtLine(LINE_H),
 ];
 
 main(args) => runIsolateTests(args, tests,