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";
-
+/**
+ * Thrown when an isolate cannot be created.
+ */
 class IsolateSpawnException implements Exception {
   const IsolateSpawnException(String this._s);
   String toString() => "IsolateSpawnException: '$_s'";
   final String _s;
 }
 
 /**
- * The initial ReceivePort available by default for this isolate.
+ * This class should be part of dart:isolate.
  *
- * 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.

[Søren Gjesse, 2013/09/27 07:04:11]

Maybe say that instances of Capability are opaque objects and from the instance
itself it is not possible to see which permission on which isolate it grants.

[floitsch, 2013/10/12 17:04:38]

Done.

  */
-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.
  *
- * 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 {

[kasperl, 2013/09/27 07:42:19]

Not sure you're going to need this class. It feels like an abstraction that
really isn't needed -- could it be part of the Isolate class instead?

[floitsch, 2013/10/12 17:04:38]

Yes, but I think there will be many Permissions.

I will write another version that doesn't use permissions yet.

+  final Symbol _permission;
+  const Permission(this._permission);
+

[Søren Gjesse, 2013/09/27 07:04:11]

FULL_CONTROL premission?

[floitsch, 2013/10/12 17:04:38]

Done.

+  // Mirror-modify basically allows everything, since we can just modify the
+  // program to our liking.
+  static const MIRROR_MODIFY = const Permission(#mirror_modify);
+  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].

[Søren Gjesse, 2013/09/27 07:04:11]

How about the arguments here vs. message above can you also send s SendPort
here? If not then they are not that similar.

[floitsch, 2013/10/12 17:04:38]

Yes. It is crucial to be able to send Sendports.

+   */
+  static Future<Isolate> spawnUri(Uri uri, var arguments, { named, args }) {
+
+  }
+
+  /**
+   * 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.

[Søren Gjesse, 2013/09/27 07:04:11]

How immediate is this pause? Pause when returning to the event loop? Assume that
there can still be messages send from the isolate until the future completes.

[floitsch, 2013/10/12 17:04:38]

There will be different ways to pause: either at the end of the event-loop, or
immediately. They will have different capabilities.

This function was an example. When we actually implement it we will think about
the names.
Suggestions so far are:
pause / suspend. (pause waits for event-loop).
kill / terminate. (kill waits for event-loop).

+   */
+  Future pause() {}
+
+  /**
+   * Resumes the isolate.
+   */
+  Future resume() {}
+
+  Future kill() {

[Søren Gjesse, 2013/09/27 07:04:11]

Should there be a optional Capability argument to pass a Capability obtained
from somewhere else?

[floitsch, 2013/10/12 17:04:38]

Interesting idea. I will think about it.

+    // Proposed implementation:
+    return controlPort.call([_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 id uniquely identifies the spawned isolate.

[Søren Gjesse, 2013/09/27 07:04:11]

"This id"?

[floitsch, 2013/10/12 17:04:38]

Old comment. Updated.

+  final SendPort controlPort;

[Søren Gjesse, 2013/09/27 07:04:11]

I assume corresponding receive port is not directly visible but handled
internally.

[floitsch, 2013/10/12 17:04:38]

Correct.

+
+  final Map<Permission, Capability> _permissions;
+
+  /**
+   * Clones this Isolate but with limited permissions.

[Søren Gjesse, 2013/09/27 07:04:11]

By "clone" don't you mean just making a new reference (controlPort) with limited
capabilities. Not starting a new isolate.

Shouldn't this return an Isolate?

[floitsch, 2013/10/12 17:04:38]

Correct.
This is usually a way to send the permissions/capabilities to the other side
(although "Permission" is currently not serializable).
The result should therefore be a Map (or something serializable).
I have to think about how to do this without the Permission class.

+   *
+   * 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.
+ * When a SendPort sends a message to a ReceivePort the receiving end receives
+ * an instance of this class.
  */
-SendPort spawnUri(String uri) => _Isolate.spawnUri(uri);
+class IsolateMessage {

[kasperl, 2013/09/27 07:42:19]

I'm not too keen on wrapping all messages in new objects. Can we avoid that? I
think we should send "raw" messages with associated ports/capabilities.

[Ivan Posva, 2013/09/30 02:59:41]

Agree with Kasper, no extra wrapping.

[floitsch, 2013/10/12 17:04:38]

Done.

+  final message;
+  final SendPort replyPort;
+  final SendPort sourceIsolate;  // should we have this?
+
+  void reply(msg, { replyTo }) {
+    if (replyPort != null) replyPort.send(msg, replyTo: replyTo);
+  }
+
+  IsolateMessage._(this.message, this.replyPort, this.sourceIsolate);
+}
 
 /**
- * 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 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]);
+  void send(var message, { SendPort replyTo });

[kasperl, 2013/09/27 07:42:19]

I think we should drop the replyTo argument (and let higher level abstraction
add that if needed). Also, I like the idea of separating out the
ports/capabilities into another list transmitted separately.

[Ivan Posva, 2013/09/30 02:59:41]

I do not see how the replyTo port is accessed on the receiving side.

[floitsch, 2013/10/12 17:04:38]

It would have been in the IsolateMessage.
But gone now.

 
   /**
    * Sends a message to this send port and returns a [Future] of the reply.
    * Basically, this internally creates a new receive port, sends a

[Ivan Posva, 2013/09/30 02:59:41]

This internal description is too detailed and we actually might want to change
the implementation at some point as it is very wasteful with port allocation.

[floitsch, 2013/10/12 17:04:38]

call is gone.

    * 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.
+   *
+   * The [Future] extracts the message from the [IsolateMessage].

[kasperl, 2013/09/27 07:42:19]

I don't like IsolateMessage in this context either.

[floitsch, 2013/10/12 17:04:38]

call is gone.

    */
   Future call(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
+ * [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, they are
  * dispatched to the callback that has been registered on the receive port.
  *
  * A [ReceivePort] may have many [SendPort]s.
  */
-abstract class ReceivePort {
+// TODO(floitsch): should the ReceivePort be a broadcast stream or not?
+// Broadcast stream: allows multiple listeners. May drop messages if nobody
+// is listening.
+// Non broadcast stream: one listener only. Buffers messages is nobody listens.
+// I vote for non-broadcast.

[kasperl, 2013/09/27 07:42:19]

Non-broadcast gives us the option to add broadcasting on top of it at a higher
level, right?

[floitsch, 2013/10/12 17:04:38]

Non-broadcast it is.

+abstract class ReceivePort implements Stream<IsolateMessage> {
 
   /**
    * 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.
+   * Inherited from [Stream].
    */
-  void receive(void callback(var message, SendPort replyTo));
+  StreamSubscription<IsolateMessage> listen(
+      void callback(IsolateMessage message),
+      { void onError(e) ,
+        void onDone(),
+        bool cancelOnError });
 
   /**
    * 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].
+   * Returns a send port that sends to this receive port.
    */
-  SendPort toSendPort();
-
+  SendPort get sendPort;
 }
 
 /**
  * [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 {
@@ skipping 23 matching lines @@
   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  ")}';
   }
 }