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

runtime/lib/async_patch.dart

Index: runtime/lib/async_patch.dart
diff --git a/runtime/lib/async_patch.dart b/runtime/lib/async_patch.dart
index 0dd99e01784cedaf278b890cba31687d2e708cde..2823b73460dda50feb67f9ad64d271a2719a406f 100644
--- a/runtime/lib/async_patch.dart
+++ b/runtime/lib/async_patch.dart
@@ -45,7 +45,10 @@ Function _asyncErrorWrapperHelper(continuation) {
 ///
 /// 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) {
@@ -59,9 +62,20 @@ Future _awaitHelper(
   //
   // 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 {
@@ -73,7 +87,14 @@ class _AsyncStarStreamController {
   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;
@@ -194,10 +215,23 @@ class _AsyncStarStreamController {
 
 @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";