Track the 'awaiter return' call stack use it to detect uncaught exceptions in async functions

runtime/lib/async_patch.dart

 // Copyright (c) 2015, 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.
 
 import "dart:_internal" hide Symbol;
 
 // Equivalent of calling FATAL from C++ code.
 _fatal(msg) native "DartAsync_fatal";
 
 // We need to pass the value as first argument and leave the second and third
@@ skipping 27 matching lines @@
   if (Zone.current == Zone.ROOT) return errorCallback;
   return Zone.current.registerBinaryCallback(errorCallback);
 }
 
 /// Registers the [thenCallback] and [errorCallback] on the given [object].
 ///
 /// If [object] is not a future, then it is wrapped into one.
 ///
 /// Returns the result of registering with `.then`.
 Future _awaitHelper(
-    var object, Function thenCallback, Function errorCallback) {
+    var object,
+    Function thenCallback,
+    Function errorCallback,
+    var awaiter) {
   if (object is! Future) {
     object = new _Future().._setValue(object);
   } else if (object is! _Future) {
     return object.then(thenCallback, onError: errorCallback);
   }
   // `object` is a `_Future`.
   //
   // Since the callbacks have been registered in the current zone (see
   // [_asyncThenWrapperHelper] and [_asyncErrorWrapperHelper]), we can avoid
   // another registration and directly invoke the no-zone-registration `.then`.
   //
   // We can only do this for our internal futures (the default implementation of
   // all futures that are constructed by the `dart:async` library).
+  object._awaiter = awaiter;
   return object._thenNoZoneRegistration(thenCallback, errorCallback);
 }
 
+// Called as part of the 'await for (...)' construct. Registers the
+// awaiter on the stream.
+void _asyncStarListenHelper(var object, var awaiter) {
+  if (object is! _StreamImpl) {
+    return;
+  }
+  // `object` is a `_StreamImpl`.
+  object._awaiter = awaiter;
+}
+
 // _AsyncStarStreamController is used by the compiler to implement
 // async* generator functions.
 class _AsyncStarStreamController {
   StreamController controller;
   Function asyncStarBody;
   bool isAdding = false;
   bool onListenReceived = false;
   bool isScheduled = false;
   bool isSuspendedAtYield = false;
   Completer cancellationCompleter = null;
 
-  Stream get stream => controller.stream;
+  Stream get stream {
+    Stream local = controller.stream;
+    if (local is! _StreamImpl) {
+      return local;
+    }
+    local._generator = asyncStarBody;
+    return local;
+  }
 
   void runBody() {
     isScheduled = false;
     isSuspendedAtYield = false;
     asyncStarBody();
   }
 
   void scheduleGenerator() {
     if (isScheduled || controller.isPaused || isAdding) {
       return;
@@ skipping 100 matching lines @@
       if (isSuspendedAtYield) {
         scheduleGenerator();
       }
     }
     return cancellationCompleter.future;
   }
 }
 
 @patch void _rethrow(Object error, StackTrace stackTrace) native "Async_rethrow";
 
+@patch class _Future<T> {
+  /// The closure implementing the async[*]-body that is `await`ing this future.
+  Function _awaiter;
+}
+
+@patch class _StreamImpl<T> {
+  /// The closure implementing the async[*]-body that is `await`ing this future.
+  Function _awaiter;
+  /// The closure implementing the async-generator body that is creating events
+  /// for this stream.
+  Function _generator;
+}
 
 /// Returns a [StackTrace] object containing the synchronous prefix for this
 /// asynchronous method.
-Object _asyncStackTraceHelper() native "StackTrace_asyncStackTraceHelper";
+Object _asyncStackTraceHelper()
+    native "StackTrace_asyncStackTraceHelper";
 
 void _clearAsyncThreadStackTrace()
     native "StackTrace_clearAsyncThreadStackTrace";
 
 void _setAsyncThreadStackTrace(StackTrace stackTrace) native
     "StackTrace_setAsyncThreadStackTrace";