Proposal for new Isolate library.

sdk/lib/isolate/isolate.dart

 // Copyright (c) 2012, 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.
 
 /**
  * Concurrent programming using _isolates_:
  * independent workers that are similar to threads
  * but don't share memory,
  * communicating only via messages.
  *
  * See also:
  * [dart:isolate - Concurrency with Isolates](https://www.dartlang.org/docs/dart-up-and-running/contents/ch03.html#ch03-dartisolate---concurrency-with-isolates)
  * in the library tour.
  */
 library dart.isolate;
 
 import "dart:async";
 
-part "isolate_stream.dart";
-
+/**

[kasperl, 2013/10/14 06:58:12]

Do we need special Isolate-specific exceptions? Are they really that beneficial?

[floitsch, 2013/10/14 17:22:40]

I think it is good style to have specialized exceptions.

+ * Thrown when an isolate cannot be created.
+ */
 class IsolateSpawnException implements Exception {

[Lasse Reichstein Nielsen, 2013/10/14 07:13:45]

Drop implementing Exception. It doesn't help anything anyway.

What causes can there be for an Isolate not to be created? Can we enumerate
them, and keep the reason in the exception, so the user knows what to do about
it.

If the user cannot do anything, consider making it an error.

[floitsch, 2013/10/14 17:22:40]

I would like that yes.
possible reasons:
- url does not exist.
- error during compilation of main (or reachable code).
- out of resources (like too many running isolates).
in these cases it is an exceptional case. In other words: the user is using the
API the right way and has no way to check that it won't throw an exception.

   const IsolateSpawnException(String this._s);

[kasperl, 2013/10/14 06:58:12]

Remove String here.

[floitsch, 2013/10/14 17:22:40]

Done.

   String toString() => "IsolateSpawnException: '$_s'";
   final String _s;
 }
 
 /**
- * The initial ReceivePort available by default for this isolate.
+ * This class should be part of dart:isolate.

[Lasse Reichstein Nielsen, 2013/10/14 07:13:45]

This sentence can be removed now.

[floitsch, 2013/10/14 17:22:40]

Done.

  *
- * This ReceivePort is created automatically
- * and is commonly used to establish
- * the first communication between isolates.
- * (See [spawnFunction] and [spawnUri].)
+ * A capability is unique among all isolates and can be shared with
+ * other isolates.
+ *
+ * Capabilities are opaque objects. They do not expose what they can be used
+ * for.
  */
-ReceivePort get port => _Isolate.port;
+class Capability {
+}
 
 /**
- * Creates and spawns an isolate
- * that shares the same code as the current isolate,
- * but that starts from the specified function.
+ * An enum that enumeratos permissions that are required to control isolates.

[Lasse Reichstein Nielsen, 2013/10/14 07:13:45]

"enumerates"

[floitsch, 2013/10/14 17:22:40]

Done.

  *
- * The [topLevelFunction] argument must be
- * a static top-level function or a static method that takes no
- * arguments. It is illegal to pass a function closure.
- *
- * When any isolate starts (even the main script of the application), a default
- * [ReceivePort] is created for it. This port is available from the top-level
- * getter [port] defined in this library.
- *
- * This function returns a [SendPort] derived from
- * the child isolate's default port.
- *
- * The optional [unhandledExceptionCallback] argument is invoked whenever an
- * exception inside the isolate is unhandled. It can be seen as a big
- * `try/catch` around everything that is executed inside the isolate. The
- * callback should return `true` if it was able to handle the exception.
+ * Some control messages require capabilities to be accepted by the isolate.
+ * The [Permission] enum gives these capabilities names.
  */
-SendPort spawnFunction(void topLevelFunction(),
-    [bool unhandledExceptionCallback(IsolateUnhandledException e)])
-    => _Isolate.spawnFunction(topLevelFunction, unhandledExceptionCallback);
+class Permission {
+  final Symbol _permission;
+  const Permission(this._permission);
+
+  // Mirror-modify basically allows everything, since we can just modify the
+  // program to our liking.
+  static const FULL_CONTROL = const Permission(#full_control);
+  static const MIRROR_INSPECT = const Permission(#mirror_inspect);
+  static const SHUTDOWN = const Permission(#shutdown);
+  // Allows to control resources (maybe even giving more than usual?).
+  static const RESOURCE = const Permission(#resource);
+}
+
+class Isolate {
+  /**
+   * Creates and spawns an isolate that shares the same code as the current
+   * isolate.
+   *
+   * The argument [entryPoint] specifies the entry point of the spawned
+   * isolate. It must be a static top-level function or a static method that
+   * takes no arguments. It is illegal to pass a function closure.
+   *
+   * The entryPoint function is invoked with the initial [message].
+   * Usually the initial [message] contains a [SendPort] so
+   * that the spawner and spawnee can communicate with each other.
+   */
+  static Future<Isolate> spawn(void entryPoint(message), var message,
+      { startPaused, or, other, Arguments, we, will, find, useful }) {
+  }
+
+  /**
+   * Creates and spawns an isolate that runs the code from the specified URI.
+   *
+   * The entry point of the spawned isolate is automatically set to
+   * `main`. Otherwise similar to [spawn].
+   */
+  static Future<Isolate> spawnUri(Uri uri, var arguments, { named, args }) {
+
+  }
+
+  /**

[kasperl, 2013/10/14 06:58:12]

I'd prefer to just get rid of the (public) control port and permission stuff for
now.

I'd pick a single method to implement on Isolate and remove the rest. Maybe the
nicest of the bunch is being able to listen for uncaught errors.

[floitsch, 2013/10/14 17:22:40]

That would mean that there is no way to send an isolate to another isolate.

+   * Creates a new isolate instance controlling an existing isolate.
+   *
+   * The new instance controls the isolate that is controlled by the
+   * argument [controlPort].
+   *
+   * The given [permissions] map specifies what the returned instance can
+   * do with the existing isolate.
+   */
+  Isolate.fromControlPort(SendPort controlPort,
+                          Map<Permission, Capability> permissions)
+      : _permissions = permissions,
+        controlPort = controlPort;
+
+  /**
+   * Listening to this stream will install an error handler on the isolate.
+   *
+   * The installation of the error handler is an asynchronous operation.
+   * If one wants to be sure to see all errors, one should start the isolate in
+   * a paused state, start listening to the errors, and then only resume the
+   * other isolate.
+   */
+  Stream uncaughtErrors;
+
+  /**
+   * Pauses the isolate.
+   */
+  Future pause() {}
+
+  /**
+   * Resumes the isolate.
+   */
+  Future resume() {}
+
+  Future kill() {
+    // Proposed implementation:
+    return controlPort.send([_someInternalObjectIdentifyingKillAction,
+                             _permissions[Permission.SHUTDOWN]]);
+  }
+
+  /**
+   * Since this is an asynchronous operation the state might change before
+   * we get a chance to act on it (if someone else has the capability to
+   * resume the isolate).
+   */
+  Future<bool> queryIsPaused() {}
+
+  /// This port uniquely identifies the spawned isolate.
+  final SendPort controlPort;
+
+  final Map<Permission, Capability> _permissions;
+
+  /**
+   * Clones this Isolate but with limited permissions.
+   *
+   * When handing out isolates, one should always
+   */
+  Map<Permission, Capability> extractPermissions(
+      Iterable<Permission> permissions) {
+  }
+}
 
 /**
- * Creates and spawns an isolate that runs the code from the specified URI.
- *
- * As with [spawnFunction],
- * the child isolate has a default [ReceivePort],
- * and this function returns a [SendPort] derived from it.
- */
-SendPort spawnUri(String uri) => _Isolate.spawnUri(uri);
-
-/**
- * Together with [ReceivePort],
- * the only means of communication between isolates.
+ * Sends messages to its [ReceivePort]s.
  *
  * [SendPort]s are created from [ReceivePort]s. Any message sent through
  * a [SendPort] is delivered to its respective [ReceivePort]. There might be
  * many [SendPort]s for the same [ReceivePort].
  *
  * [SendPort]s can be transmitted to other isolates.
  */
 abstract class SendPort {
 
   /**
    * Sends an asynchronous [message] to this send port. The message is copied to
-   * the receiving isolate. If specified, the [replyTo] port will be provided to
-   * the receiver to facilitate exchanging sequences of messages.
+   * the receiving isolate.
    *
    * The content of [message] can be: primitive values (null, num, bool, double,
    * String), instances of [SendPort], and lists and maps whose elements are any
    * of these. List and maps are also allowed to be cyclic.
    *
    * In the special circumstances when two isolates share the same code and are
    * running in the same process (e.g. isolates created via [spawnFunction]), it
    * is also possible to send object instances (which would be copied in the
    * process). This is currently only supported by the dartvm.  For now, the
    * dart2js compiler only supports the restricted messages described above.
-   *
-   * Deprecation note: it is no longer valid to transmit a [ReceivePort] in a
-   * message. Previously they were translated to the corresponding send port
-   * before being transmitted.
    */
-  void send(var message, [SendPort replyTo]);
-
-  /**
-   * Sends a message to this send port and returns a [Future] of the reply.
-   * Basically, this internally creates a new receive port, sends a
-   * message to this send port with replyTo set to such receive port, and, when
-   * a reply is received, it closes the receive port and completes the returned
-   * future.
-   */
-  Future call(var message);
+  void send(var message);
 
   /**
    * Tests whether [other] is a [SendPort] pointing to the same
    * [ReceivePort] as this one.
    */
   bool operator==(var other);
 
   /**
    * Returns an immutable hash code for this send port that is
    * consistent with the == operator.
    */
   int get hashCode;
-
-}
-
-/**
- * Together with [SendPort], the only means of
- * communication between isolates.
- *
- * [ReceivePort]s have a [:toSendPort:] method
- * which returns a [SendPort]. Any message that is sent through this [SendPort]
- * is delivered to the [ReceivePort] it has been created from. There, they are
- * dispatched to the callback that has been registered on the receive port.
- *
- * A [ReceivePort] may have many [SendPort]s.
- */
-abstract class ReceivePort {
-
-  /**
-   * Opens a long-lived port for receiving messages. The returned port
-   * must be explicitly closed through [ReceivePort.close].
-   */
-  external factory ReceivePort();
-
-  /**
-   * Sets up a callback function for receiving pending or future
-   * messages on this receive port.
-   */
-  void receive(void callback(var message, SendPort replyTo));
-
-  /**
-   * Closes this receive port immediately. Pending messages will not
-   * be processed and it is impossible to re-open the port. Single-shot
-   * reply ports, such as those created through [SendPort.call], are
-   * automatically closed when the reply has been received. Multiple
-   * invocations of [close] are allowed but ignored.
-   */
-  void close();
-
-  /**
-   * Creates a new send port that sends to this receive port. It is legal to
-   * create several [SendPort]s from the same [ReceivePort].
-   */
-  SendPort toSendPort();
-
-}
-
-/**
- * [SendPortSync]s are created from [ReceivePortSync]s. Any message sent through
- * a [SendPortSync] is delivered to its respective [ReceivePortSync]. There
- * might be many [SendPortSync]s for the same [ReceivePortSync].
- *
- * [SendPortSync]s can be transmitted to other isolates.
- */
-abstract class SendPortSync {
-  /**
-   * Sends a synchronous message to this send port and returns the result.
-   */
-  callSync(var message);
-
-  /**
-   * Tests whether [other] is a [SendPortSync] pointing to the same
-   * [ReceivePortSync] as this one.
-   */
-  bool operator==(var other);
-
-  /**
-   * Returns an immutable hash code for this send port that is
-   * consistent with the == operator.
-   */
-  int get hashCode;
 }
 
+/**
+ * Together with [SendPort], the only means of communication between isolates.
+ *
+ * [ReceivePort]s have a `sendport` getter which returns a [SendPort].
+ * Any message that is sent through this [SendPort]
+ * is delivered to the [ReceivePort] it has been created from. There, the
+ * message is dispatched to its listener.
+ *
+ * A [ReceivePort] is a non-broadcast stream. This means that it buffers
+ * incoming messages until a listener is registered. Only one listener can
+ * receive messages. See [Stream.asBroadcastStream] for transforming the port
+ * to a broadcast stream.
+ *
+ * A [ReceivePort] may have many [SendPort]s.
+ */
+abstract class ReceivePort implements Stream {
+
+  /**
+   * Opens a long-lived port for receiving messages.
+   *
+   * A [ReceivePort] is a non-broadcast stream. This means that it buffers
+   * incoming messages until a listener is registered. Only one listener can
+   * receive messages. See [Stream.asBroadcastStream] for transforming the port
+   * to a broadcast stream.
+   *
+   * A receive port is closed by canceling its subscription.
+   */
+  external factory ReceivePort();
+
+  /**
+   * Inherited from [Stream].
+   *
+   * Note that all named arguments are ignored since a ReceivePort will never
+   * receive an error, or done message.
+   */
+  StreamSubscription listen(void callback(var message),
+                            { Function onError,
+                              void onDone(),
+                              bool cancelOnError });
+
+  /**
+   * Returns a send port that sends to this receive port.
+   */
+  SendPort get sendPort;
+}
+
 // The VM doesn't support accessing external globals in the same library. We
 // therefore create this wrapper class.
 // TODO(6997): Don't go through static class for external variables.
 abstract class _Isolate {
   external static ReceivePort get port;
   external static SendPort spawnFunction(void topLevelFunction(),
     [bool unhandledExceptionCallback(IsolateUnhandledException e)]);
   external static SendPort spawnUri(String uri);
 }
 
 /**
  * Wraps unhandled exceptions thrown during isolate execution. It is
  * used to show both the error message and the stack trace for unhandled
  * exceptions.
  */
+// TODO(floitsch): probably going to remove and replace with something else.
 class IsolateUnhandledException implements Exception {
   /** Message being handled when exception occurred. */
   final message;
 
   /** Wrapped exception. */
   final source;
 
   /** Trace for the wrapped exception. */
   final Object stackTrace;
 
   const IsolateUnhandledException(this.message, this.source, this.stackTrace);
 
   String toString() {
     return 'IsolateUnhandledException: exception while handling message: '
         '${message} \n  '
         '${source.toString().replaceAll("\n", "\n  ")}\n'
         'original stack trace:\n  '
         '${stackTrace.toString().replaceAll("\n","\n  ")}';
   }
 }