[vm] More efficient 'await' of not Future and completed _Future

When awaiting a value which is not a Future or a completed
built-in _Future, 'await' implementation can bypass heavyweight
_Future/_FutureListener machinery and schedule micro-tasks directly.

Benchmarks:
JIT, x64:
AsyncLiveVars.* +46-54% (bigger is better)
Calls.AwaitAsyncCall -46% (less is better)
Calls.AwaitAsyncCallInstanceTargetPolymorphic -46%
Calls.AwaitAsyncCallClosureTargetPolymorphic -45%
Calls.AwaitFutureOrCallInstanceTargetPolymorphicManyAwaits -45%
Calls.AwaitFutureOrCall -60%
Calls.AwaitFutureOrCallInstanceTargetPolymorphic -60%
Calls.AwaitFutureOrCallClosureTargetPolymorphic -59%

JIT, ia32:
AsyncLiveVars.* +43-52% (bigger is better)
Calls.AwaitAsyncCall -42% (less is better)
Calls.AwaitAsyncCallInstanceTargetPolymorphic -42%
Calls.AwaitAsyncCallClosureTargetPolymorphic -41%
Calls.AwaitFutureOrCallInstanceTargetPolymorphicManyAwaits -39%
Calls.AwaitFutureOrCall -55%
Calls.AwaitFutureOrCallInstanceTargetPolymorphic -54%
Calls.AwaitFutureOrCallClosureTargetPolymorphic -53%

JIT, arm:
AsyncLiveVars.* +64-71% (bigger is better)
Calls.AwaitAsyncCallInstanceTargetPolymorphic -51% (less is better)
Calls.AwaitAsyncCallClosureTargetPolymorphic -47%
Calls.AwaitFutureOrCallInstanceTargetPolymorphicManyAwaits -48%
Calls.AwaitFutureOrCall -64%
Calls.AwaitFutureOrCallInstanceTargetPolymorphic -64%
Calls.AwaitFutureOrCallClosureTargetPolymorphic -59%

JIT, arm64:
AsyncLiveVars.* +65-78% (bigger is better)
Calls.AwaitAsyncCall -51% (less is better)
Calls.AwaitAsyncCallInstanceTargetPolymorphic -51%
Calls.AwaitAsyncCallClosureTargetPolymorphic -50%
Calls.AwaitFutureOrCallInstanceTargetPolymorphicManyAwaits -49%
Calls.AwaitFutureOrCall -69%
Calls.AwaitFutureOrCallInstanceTargetPolymorphic -68%
Calls.AwaitFutureOrCallClosureTargetPolymorphic -67%

AOT, x64:
AsyncLiveVars.* +55-61% (bigger is better)
Calls.AwaitAsyncCall -47% (less is better)
Calls.AwaitAsyncCallInstanceTargetPolymorphic -46%
Calls.AwaitAsyncCallClosureTargetPolymorphic -47%
Calls.AwaitFutureOrCallInstanceTargetPolymorphicManyAwaits -46%
Calls.AwaitFutureOrCall -59%
Calls.AwaitFutureOrCallInstanceTargetPolymorphic -59%
Calls.AwaitFutureOrCallClosureTargetPolymorphic -58%

AOT, arm:
AsyncLiveVars.* 54-66% (bigger is better)
Calls.AwaitAsyncCall -46-51% (less is better)
Calls.AwaitAsyncCallInstanceTargetPolymorphic -46-50%
Calls.AwaitAsyncCallClosureTargetPolymorphic -46-52%
Calls.AwaitFutureOrCallInstanceTargetPolymorphicManyAwaits -45-50%
Calls.AwaitFutureOrCall -63-68%
Calls.AwaitFutureOrCallInstanceTargetPolymorphic -63-66%
Calls.AwaitFutureOrCallClosureTargetPolymorphic -63-67%

AOT, arm64:
AsyncLiveVars.* +53-66% (bigger is better)
Calls.AwaitAsyncCall -50-51% (less is better)
Calls.AwaitAsyncCallInstanceTargetPolymorphic -50%
Calls.AwaitAsyncCallClosureTargetPolymorphic -50-51%
Calls.AwaitFutureOrCallInstanceTargetPolymorphicManyAwaits -49-50%
Calls.AwaitFutureOrCall -66-68%
Calls.AwaitFutureOrCallInstanceTargetPolymorphic -66-68%
Calls.AwaitFutureOrCallClosureTargetPolymorphic -63-67%

TEST=ci
Issue: #48378

Change-Id: I65e3702fcd816ee3fee876ff442b9887c035b1ec
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/243102
Reviewed-by: Lasse Nielsen <[email protected]>
Commit-Queue: Alexander Markov <[email protected]>

Author: alexmarkov

pkg/vm_service/test/get_stack_test.dart

@@ -95,7 +95,7 @@ final tests = <IsolateTest>[
   (VmService service, IsolateRef isolateRef) async {
     final result = await service.getStack(isolateRef.id!);
 
-    expect(result.frames, hasLength(10));
+    expect(result.frames, hasLength(6));
     expect(result.asyncCausalFrames, hasLength(26));
     expect(result.awaiterFrames, hasLength(13));
 
@@ -105,10 +105,6 @@ final tests = <IsolateTest>[
       [equals('Regular'), anything], // Internal mech. ..
       [equals('Regular'), anything],
       [equals('Regular'), anything],
-      [equals('Regular'), anything],
-      [equals('Regular'), anything],
-      [equals('Regular'), anything],
-      [equals('Regular'), anything],
       [equals('Regular'), endsWith(' _RawReceivePortImpl._handleMessage')],
     ]);
 

runtime/observatory/tests/service/get_stack_limit_rpc_test.dart

@@ -51,7 +51,7 @@ var tests = <IsolateTest>[
     var frames = stack['frames'];
     var asyncFrames = stack['asyncCausalFrames'];
     var awaiterFrames = stack['awaiterFrames'];
-    expect(frames.length, greaterThanOrEqualTo(20));
+    expect(frames.length, greaterThanOrEqualTo(12));
     expect(asyncFrames.length, greaterThan(frames.length));
     expect(awaiterFrames.length, greaterThan(frames.length));
     expect(stack['truncated'], false);

runtime/observatory_2/tests/service_2/get_stack_limit_rpc_test.dart

@@ -51,7 +51,7 @@ var tests = <IsolateTest>[
     var frames = stack['frames'];
     var asyncFrames = stack['asyncCausalFrames'];
     var awaiterFrames = stack['awaiterFrames'];
-    expect(frames.length, greaterThanOrEqualTo(20));
+    expect(frames.length, greaterThanOrEqualTo(12));
     expect(asyncFrames.length, greaterThan(frames.length));
     expect(awaiterFrames.length, greaterThan(frames.length));
     expect(stack['truncated'], false);

sdk/lib/_internal/vm/lib/async_patch.dart

@@ -372,15 +372,69 @@ class _SuspendState {
     }
   }
 
+  @pragma("vm:invisible")
+  @pragma("vm:prefer-inline")
+  void _awaitCompletedFuture(_Future future) {
+    assert(future._isComplete);
+    final zone = Zone._current;
+    if (future._hasError) {
+      @pragma("vm:invisible")
+      void run() {
+        final AsyncError asyncError =
+            unsafeCast<AsyncError>(future._resultOrListeners);
+        zone.runBinary(
+            unsafeCast<dynamic Function(Object, StackTrace)>(_errorCallback),
+            asyncError.error,
+            asyncError.stackTrace);
+      }
+
+      zone.scheduleMicrotask(run);
+    } else {
+      @pragma("vm:invisible")
+      void run() {
+        zone.runUnary(unsafeCast<dynamic Function(dynamic)>(_thenCallback),
+            future._resultOrListeners);
+      }
+
+      zone.scheduleMicrotask(run);
+    }
+  }
+
+  @pragma("vm:invisible")
+  @pragma("vm:prefer-inline")
+  void _awaitNotFuture(Object? object) {
+    final zone = Zone._current;
+    @pragma("vm:invisible")
+    void run() {
+      zone.runUnary(
+          unsafeCast<dynamic Function(dynamic)>(_thenCallback), object);
+    }
+
+    zone.scheduleMicrotask(run);
+  }
+
   @pragma("vm:entry-point", "call")
   @pragma("vm:invisible")
   Object? _await(Object? object) {
-    if (_trace) print('_awaitAsync (object=$object)');
+    if (_trace) print('_await (object=$object)');
     if (_thenCallback == null) {
       _createAsyncCallbacks();
     }
-    _awaitHelper(object, unsafeCast<dynamic Function(dynamic)>(_thenCallback),
-        unsafeCast<dynamic Function(Object, StackTrace)>(_errorCallback));
+    if (object is _Future) {
+      if (object._isComplete) {
+        _awaitCompletedFuture(object);
+      } else {
+        object._thenAwait<dynamic>(
+            unsafeCast<dynamic Function(dynamic)>(_thenCallback),
+            unsafeCast<dynamic Function(Object, StackTrace)>(_errorCallback));
+      }
+    } else if (object is! Future) {
+      _awaitNotFuture(object);
+    } else {
+      object.then(unsafeCast<dynamic Function(dynamic)>(_thenCallback),
+          onError:
+              unsafeCast<dynamic Function(Object, StackTrace)>(_errorCallback));
+    }
     return _functionData;
   }