Add an error listener on isolates.

While any error listener is attached to an isolate, uncaught errors are reported to the listener through his supplied sendPort.

The current format for the error is
  [error.toString(), stack == null ? null : stack.toString()]
This is wrapped on the receiving side.

The isolate exposes the errors as a broadcast stream of error events.

R=floitsch@google.com

Review URL: https://codereview.chromium.org//169703002

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@36290 260f80e4-7a28-3924-810f-c04153c831b5

sdk/lib/_internal/lib/isolate_helper.dart

@@ -89,7 +89,6 @@ void startRootIsolate(entry, args) {
   // isolate automatically we try to give them a reasonable context to live in
   // by having a "default" isolate (the first one created).
   _globalState.currentContext = rootContext;
-
   if (entry is _MainFunctionArgs) {
     rootContext.eval(() { entry(args); });
   } else if (entry is _MainFunctionArgsMessage) {
@@ -271,6 +270,13 @@ class _IsolateContext implements IsolateContext {
   final Capability pauseCapability = new Capability();
   final Capability terminateCapability = new Capability();  // License to kill.
 
+  /// Boolean flag set when the initial method of the isolate has been executed.
+  ///
+  /// Used to avoid considering the isolate dead when it has no open
+  /// receive ports and no scheduled timers, because it hasn't had time to
+  /// create them yet.
+  bool initialized = false;
+
   // TODO(lrn): Store these in single "PauseState" object, so they don't take
   // up as much room when not pausing.
   bool isPaused = false;
@@ -290,11 +296,11 @@ class _IsolateContext implements IsolateContext {
   var _scheduledControlEvents;
   bool _isExecutingEvent = false;
 
-  /** Whether errors are considered fatal. */
-  // This doesn't do anything yet. We need to be able to catch uncaught errors
-  // (oxymoronically) in order to take lethal action. This is waiting for the
-  // same change as the uncaught error listeners.
-  bool errorsAreFatal = false;
+  /** Whether uncaught errors are considered fatal. */
+  bool errorsAreFatal = true;
+
+  // Set of ports that listen to uncaught errors.
+  Set<SendPort> errorPorts = new Set();
 
   _IsolateContext() {
     this.registerWeak(controlPort._id, controlPort);
@@ -379,6 +385,40 @@ class _IsolateContext implements IsolateContext {
     _scheduledControlEvents.addLast(kill);
   }
 
+  void addErrorListener(SendPort port) {
+    errorPorts.add(port);
+  }
+
+  void removeErrorListener(SendPort port) {
+    errorPorts.remove(port);
+  }
+
+  /** Function called with an uncaught error. */
+  void handleUncaughtError(error, StackTrace stackTrace) {
+    // Just print the error if there is no error listener registered.
+    if (errorPorts.isEmpty) {
+      // An uncaught error in the root isolate will terminate the program?
+      if (errorsAreFatal && identical(this, _globalState.rootContext)) {
+        // The error will be rethrown to reach the global scope, so
+        // don't print it.
+        return;
+      }
+      if (JS('bool', '#.console != null && '
+                     'typeof #.console.error == "function"',
+                     globalThis, globalThis)) {
+        JS('void', '#.console.error(#, #)', globalThis, error, stackTrace);
+      } else {
+        print(error);
+        if (stackTrace != null) print(stackTrace);
+      }
+      return;
+    }
+    List message = new List(2)
+        ..[0] = error.toString()
+        ..[1] = (stackTrace == null) ? null : stackTrace.toString();
+    for (SendPort port in errorPorts) port.send(message);
+  }
+
   /**
    * Run [code] in the context of the isolate represented by [this].
    */
@@ -390,6 +430,15 @@ class _IsolateContext implements IsolateContext {
     _isExecutingEvent = true;
     try {
       result = code();
+    } catch (e, s) {
+      handleUncaughtError(e, s);
+      if (errorsAreFatal) {
+        kill();
+        // An uncaught error in the root context terminates all isolates.
+        if (identical(this, _globalState.rootContext)) {
+          rethrow;
+        }
+      }
     } finally {
       _isExecutingEvent = false;
       _globalState.currentContext = old;
@@ -437,6 +486,12 @@ class _IsolateContext implements IsolateContext {
       case "kill":
         handleKill(message[1], message[2]);
         break;
+      case "getErrors":
+        addErrorListener(message[1]);
+        break;
+      case "stopErrors":
+        removeErrorListener(message[1]);
+        break;
       default:
     }
   }
@@ -468,7 +523,7 @@ class _IsolateContext implements IsolateContext {
   }
 
   void _updateGlobalState() {
-    if (ports.length - weakPorts.length > 0 || isPaused) {
+    if (ports.length - weakPorts.length > 0 || isPaused || !initialized) {
       _globalState.isolates[id] = this; // indicate this isolate is active
     } else {
       kill();
@@ -490,6 +545,7 @@ class _IsolateContext implements IsolateContext {
     ports.clear();
     weakPorts.clear();
     _globalState.isolates.remove(id); // indicate this isolate is not active
+    errorPorts.clear();
     if (doneHandlers != null) {
       for (SendPort port in doneHandlers) {
         port.send(null);
@@ -920,6 +976,7 @@ class IsolateNatives {
                   context.terminateCapability]);
 
     void runStartFunction() {
+      context.initialized = true;
       if (!isSpawnUri) {
         topLevel(message);
       } else if (topLevel is _MainFunctionArgsMessage) {

sdk/lib/isolate/isolate.dart

@@ -47,6 +47,7 @@ class Isolate {
    * Capability granting the ability to pause the isolate.
    */
   final Capability pauseCapability;
+
   /**
    * Capability granting the ability to terminate the isolate.
    */
@@ -298,6 +299,80 @@ class Isolate {
         ..[2] = pingType;
     controlPort.send(message);
   }
+
+  /**
+   * Requests that uncaught errors of the isolate are sent back to [port].
+   *
+   * The errors are sent back as two elements lists.
+   * The first element is a `String` representation of the error, usually
+   * created by calling `toString` on the error.
+   * The second element is a `String` representation of an accompanying
+   * stack trace, or `null` if no stack trace was provided.
+   *
+   * Listening using the same port more than once does nothing. It will only
+   * get each error once.
+   */
+  void addErrorListener(SendPort port) {
+    var message = new List(2)
+        ..[0] = "getErrors"
+        ..[1] = port;
+    controlPort.send(message);
+  }
+
+  /**
+   * Stop listening for uncaught errors through [port].
+   *
+   * The `port` should be a port that is listening for errors through
+   * [addErrorListener]. This call requests that the isolate stops sending
+   * errors on the port.
+   *
+   * If the same port has been passed via `addErrorListener` more than once,
+   * only one call to `removeErrorListener` is needed to stop it from receiving
+   * errors.
+   *
+   * Closing the receive port at the end of the send port will not stop the
+   * isolate from sending errors, they are just going to be lost.
+   */
+  void removeErrorListener(SendPort port) {
+    var message = new List(2)
+        ..[0] = "stopErrors"
+        ..[1] = port;
+    controlPort.send(message);
+  }
+
+  /**
+   * Returns a broadcast stream of uncaught errors from the isolate.
+   *
+   * Each error is provided as an error event on the stream.
+   *
+   * The actual error object and stackTraces will not necessarily
+   * be the same object types as in the actual isolate, but they will
+   * always have the same [Object.toString] result.
+   *
+   * This stream is based on [addErrorListener] and [removeErrorListener].
+   */
+  Stream get errors {
+    StreamController controller;
+    RawReceivePort port;
+    void handleError(message) {
+      String errorDescription = message[0];
+      String stackDescription = message[1];
+      var error = new RemoteError(errorDescription, stackDescription);
+      controller.addError(error, error.stackTrace);
+    }
+    controller = new StreamController.broadcast(
+        sync: true,
+        onListen: () {
+          port = new RawReceivePort(handleError);
+          this.addErrorListener(port.sendPort);
+        },
+        onCancel: () {
+          this.removeErrorListener(port.sendPort);
+          port.close();
+          port = null;
+        });
+    return controller.stream;
+  }
 }
 
 /**
@@ -464,3 +539,24 @@ class _IsolateUnhandledException implements Exception {
         '${stackTrace.toString().replaceAll("\n","\n  ")}';
   }
 }
+
+/**
+ * Description of an error from another isolate.
+ *
+ * This error has the same `toString()` and `stackTrace.toString()` behavior
+ * as the original error, but has no other features of the original error.
+ */
+class RemoteError implements Error {
+  final String _description;
+  final StackTrace stackTrace;
+  RemoteError(String description, String stackDescription)
+      : _description = description,
+        stackTrace = new _RemoteStackTrace(stackDescription);
+  String toString() => _description;
+}
+
+class _RemoteStackTrace implements StackTrace {
+  String _trace;
+  _RemoteStackTrace(this._trace);
+  String toString() => _trace;
+}

tests/isolate/handle_error2_test.dart

@@ -0,0 +1,89 @@
+// Copyright (c) 2014, 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.
+
+library handle_error_test;
+
+import "dart:isolate";
+import "dart:async";
+import "package:async_helper/async_helper.dart";
+import "package:expect/expect.dart";
+
+isomain1(replyPort) {
+  RawReceivePort port = new RawReceivePort();
+  port.handler = (v) {
+    switch (v) {
+      case 0:
+        replyPort.send(42);
+        break;
+      case 1:
+        throw new ArgumentError("whoops");
+      case 2:
+        throw new RangeError.value(37);
+      case 3:
+        port.close();
+    }
+  };
+  replyPort.send(port.sendPort);
+}
+
+/// Do Isolate.spawn(entry) and get a sendPort from the isolate that it
+/// expects commands on.
+/// The isolate has errors set to non-fatal.
+/// Returns a list of `[isolate, commandPort]` in a future.
+Future spawn(entry) {
+  ReceivePort reply = new ReceivePort();
+  Future isolate = Isolate.spawn(entry, reply.sendPort, paused: true);
+  return isolate.then((Isolate isolate) {
+    isolate.setErrorsFatal(false);
+    isolate.resume(isolate.pauseCapability);
+    Future result = reply.first.then((sendPort) {
+      return [isolate, sendPort];
+    });
+    return result;
+  });
+}
+
+main(){
+  asyncStart();
+  RawReceivePort reply = new RawReceivePort(null);
+  RawReceivePort reply2 = new RawReceivePort(null);
+  // Create two isolates waiting for commands, with errors non-fatal.
+  Future iso1 = spawn(isomain1);
+  Future iso2 = spawn(isomain1);
+  Future.wait([iso1, iso2]).then((l) {
+    var isolate1 = l[0][0];
+    var sendPort1 = l[0][1];
+    var isolate2 = l[1][0];
+    var sendPort2 = l[1][1];
+    Stream errors = isolate1.errors;  // Broadcast stream, never a done message.
+    int state = 1;
+    var subscription;
+    subscription = errors.listen(null, onError: (error, stack) {
+      switch (state) {
+        case 1:
+          Expect.equals(new ArgumentError("whoops").toString(), "$error");
+          state++;
+          break;
+        case 2:
+          Expect.equals(new RangeError.value(37).toString(), "$error");
+          state++;
+          reply.close();
+          subscription.cancel();
+          asyncEnd();
+          break;
+        default:
+          throw "Bad state for error: $state: $error";
+      }
+    });
+    sendPort1.send(0);
+    sendPort2.send(0);
+    sendPort1.send(1);
+    sendPort2.send(1);
+    sendPort1.send(2);
+    sendPort2.send(2);
+    sendPort1.send(3);
+    sendPort2.send(3);
+  });
+}
+

tests/isolate/handle_error3_test.dart

@@ -0,0 +1,116 @@
+// Copyright (c) 2014, 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.
+
+library handle_error_test;
+
+import "dart:isolate";
+import "dart:async";
+import "package:async_helper/async_helper.dart";
+import "package:expect/expect.dart";
+
+isomain1(replyPort) {
+  RawReceivePort port = new RawReceivePort();
+  port.handler = (v) {
+    switch (v) {
+      case 0:
+        replyPort.send(42);
+        break;
+      case 1:
+        throw new ArgumentError("whoops");
+      case 2:
+        throw new RangeError.value(37);
+      case 3:
+        port.close();
+    }
+  };
+  replyPort.send(port.sendPort);
+}
+
+/// Do Isolate.spawn(entry) and get a sendPort from the isolate that it
+/// expects commands on.
+/// The isolate has errors set to non-fatal.
+/// Returns a list of `[isolate, commandPort]` in a future.
+Future spawn(entry) {
+  ReceivePort reply = new ReceivePort();
+  Future isolate = Isolate.spawn(entry, reply.sendPort, paused: true);
+  return isolate.then((Isolate isolate) {
+    isolate.setErrorsFatal(false);
+    isolate.resume(isolate.pauseCapability);
+    Future result = reply.first.then((sendPort) {
+      return [isolate, sendPort];
+    });
+    return result;
+  });
+}
+
+main(){
+  asyncStart();
+  asyncStart();
+  RawReceivePort reply = new RawReceivePort(null);
+  RawReceivePort reply2 = new RawReceivePort(null);
+  // Create two isolates waiting for commands, with errors non-fatal.
+  Future iso1 = spawn(isomain1);
+  Future iso2 = spawn(isomain1);
+  Future.wait([iso1, iso2]).then((l) {
+    var isolate1 = l[0][0];
+    var sendPort1 = l[0][1];
+    var isolate2 = l[1][0];
+    var sendPort2 = l[1][1];
+    // Capture errors from one isolate as stream.
+    Stream errors = isolate1.errors;  // Broadcast stream, never a done message.
+    int state = 1;
+    var subscription;
+    subscription = errors.listen(null, onError: (error, stack) {
+      switch (state) {
+        case 1:
+          Expect.equals(new ArgumentError("whoops").toString(), "$error");
+          state++;
+          break;
+        case 2:
+          Expect.equals(new RangeError.value(37).toString(), "$error");
+          state++;
+          reply.close();
+          subscription.cancel();
+          asyncEnd();
+          break;
+        default:
+          throw "Bad state for error: $state: $error";
+      }
+    });
+    // Capture errors from other isolate as raw messages.
+    RawReceivePort errorPort2 = new RawReceivePort();
+    int state2 = 1;
+    errorPort2.handler = (message) {
+      String error = message[0];
+      String stack = message[1];
+      switch (state2) {
+        case 1:
+          Expect.equals(new ArgumentError("whoops").toString(), "$error");
+          state2++;
+          break;
+        case 2:
+          Expect.equals(new RangeError.value(37).toString(), "$error");
+          state2++;
+          reply.close();
+          isolate2.removeErrorListener(errorPort2.sendPort);
+          errorPort2.close();
+          asyncEnd();
+          break;
+        default:
+          throw "Bad state-2 for error: $state: $error";
+      }
+    };
+    isolate2.addErrorListener(errorPort2.sendPort);
+
+    sendPort1.send(0);
+    sendPort2.send(0);
+    sendPort1.send(1);
+    sendPort2.send(1);
+    sendPort1.send(2);
+    sendPort2.send(2);
+    sendPort1.send(3);
+    sendPort2.send(3);
+  });
+}
+

tests/isolate/handle_error_test.dart

@@ -0,0 +1,71 @@
+// Copyright (c) 2014, 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.
+
+library handle_error_test;
+
+import "dart:isolate";
+import "dart:async";
+import "package:async_helper/async_helper.dart";
+import "package:expect/expect.dart";
+
+isomain1(replyPort) {
+  RawReceivePort port = new RawReceivePort();
+  port.handler = (v) {
+    switch (v) {
+      case 0:
+        replyPort.send(42);
+        break;
+      case 1:
+        throw new ArgumentError("whoops");
+      case 2:
+        throw new RangeError.value(37);
+      case 3:
+        port.close();
+    }
+  };
+  replyPort.send(port.sendPort);
+}
+
+main(){
+  asyncStart();
+  RawReceivePort reply = new RawReceivePort(null);
+  // Start paused so we have time to set up the error handler.
+  Isolate.spawn(isomain1, reply.sendPort, paused: true).then((Isolate isolate) {
+    isolate.setErrorsFatal(false);
+    Stream errors = isolate.errors;  // Broadcast stream, never a done message.
+    SendPort sendPort;
+    StreamSubscription subscription;
+    int state = 0;
+    reply.handler = (port) {
+      sendPort = port;
+      port.send(state);
+      reply.handler = (v) {
+        Expect.equals(0, state);
+        Expect.equals(42, v);
+        state++;
+        sendPort.send(state);
+      };
+    };
+    subscription = errors.listen(null, onError: (error, stack) {
+      switch (state) {
+        case 1:
+          Expect.equals(new ArgumentError("whoops").toString(), "$error");
+          state++;
+          sendPort.send(state);
+          break;
+        case 2:
+          Expect.equals(new RangeError.value(37).toString(), "$error");
+          state++;
+          sendPort.send(state);
+          reply.close();
+          subscription.cancel();
+          asyncEnd();
+          break;
+        default:
+          throw "Bad state for error: $state: $error";
+      }
+    });
+    isolate.resume(isolate.pauseCapability);
+  });
+}

tests/isolate/isolate.status

@@ -23,6 +23,9 @@ kill_test: Skip         # Not implemented yet, hangs.
 kill2_test: Skip        # Not implemented yet, hangs.
 kill3_test: Skip        # Not implemented yet, hangs.
 kill_self_test: Skip    # Not implemented yet, hangs.
+handle_error_test: Skip # Not implemented yet, hangs.
+handle_error2_test: Skip # Not implemented yet, hangs.
+handle_error3_test: Skip # Not implemented yet, hangs.
 
 [ $compiler == dart2js && $jscl ]
 browser/*: SkipByDesign  # Browser specific tests
@@ -32,6 +35,7 @@ pause_test: Fail  # non-zero timer not supported.
 
 [ $compiler == dart2js ]
 serialization_test: RuntimeError # Issue 1882, tries to access class TestingOnly declared in isolate_patch.dart
+isolate_throws_test/01: Fail, OK # Issue 12588. Dart2js no longer dies when an isolate throws.
 
 [ $compiler == dart2js && $runtime == ie9 ]
 browser/typed_data_message_test: Fail # Issue 12624