Very simple version of Isolates.

runtime/lib/isolate_patch.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.
 
-class _CloseToken {
-  /// This token is sent from [IsolateSink]s to [IsolateStream]s to ask them to
-  /// close themselves.
-  const _CloseToken();
+patch class ReceivePort {
+  /* patch */ factory ReceivePort() = _ReceivePortImpl;
+
+  /* patch */ factory ReceivePort.fromRawReceivePort(RawReceivePort rawPort) =
+      _ReceivePortImpl.fromRawReceivePort;
 }
 
-patch bool _isCloseToken(var object) {
-  // TODO(floitsch): can we compare against const _CloseToken()?
-  return object is _CloseToken;
-}
-
-patch class MessageBox {
-  /* patch */ MessageBox.oneShot() : this._oneShot(new ReceivePort());
-  MessageBox._oneShot(ReceivePort receivePort)
-      : stream = new IsolateStream._fromOriginalReceivePortOneShot(receivePort),
-        sink = new _IsolateSink._fromPort(receivePort.toSendPort());
-
-  /* patch */ MessageBox() : this._(new ReceivePort());
-  MessageBox._(ReceivePort receivePort)
-      : stream = new IsolateStream._fromOriginalReceivePort(receivePort),
-        sink = new _IsolateSink._fromPort(receivePort.toSendPort());
-}
-
-class _IsolateSink implements IsolateSink {
-  bool _isClosed = false;
-  final SendPort _port;
-  _IsolateSink._fromPort(this._port);
-
-  void add(dynamic message) {
-    _port.send(message);
-  }
-
-  void addError(Object errorEvent) {
-    throw new UnimplementedError("addError on isolate streams");
-  }
-
-  void close() {
-    if (_isClosed) return;
-    add(const _CloseToken());
-    _isClosed = true;
-  }
-
-  bool operator==(var other) {
-    return other is IsolateSink && _port == other._port;
-  }
-
-  int get hashCode => _port.hashCode + 499;
-}
-
-patch IsolateSink streamSpawnFunction(
-    void topLevelFunction(),
-    [bool unhandledExceptionCallback(IsolateUnhandledException e)]) {
-  SendPort sendPort = spawnFunction(topLevelFunction,
-                                    unhandledExceptionCallback);
-  return new _IsolateSink._fromPort(sendPort);
-}
-
-patch class ReceivePort {
-  /* patch */ factory ReceivePort() {
-    return new _ReceivePortImpl();
+patch class RawReceivePort {
+  /**
+   * Opens a long-lived port for receiving messages.
+   *
+   * A [RawReceivePort] is low level and does not work with [Zone]s. It
+   * can not be paused. The data-handler must be set before the first
+   * event is received.

[Lasse Reichstein Nielsen, 2013/10/25 09:42:49]

How about not making the handler optional?

You can still pass 'null' if you really, really want to (e.g., initializing a
final port field in a constructor, but setting the 
handler to an instance method).

Or making it default to something that throws, so it's not the port itself that
has to care about nulls.

[floitsch, 2013/10/25 13:11:01]

The port doesn't care. It just tries to call it. If it's null, bad luck...

Willing to discuss.

+   */
+  /* patch */ factory RawReceivePort([void handler(event)]) {
+    _RawReceivePortImpl result = new _RawReceivePortImpl();
+    result.handler = handler;
+    return result;
   }
 }
 
-class _ReceivePortImpl implements ReceivePort {
-  factory _ReceivePortImpl() native "ReceivePortImpl_factory";
+class _ReceivePortImpl extends Stream implements ReceivePort {
+  _ReceivePortImpl() : this.fromRawReceivePort(new RawReceivePort());
 
-  receive(void onMessage(var message, SendPort replyTo)) {
-    _onMessage = onMessage;
+  _ReceivePortImpl.fromRawReceivePort(this._rawPort) {
+    _controller = new StreamController(onCancel: close, sync: true);
+    _rawPort.handler = _controller.add;
   }
 
+  SendPort get sendPort {
+    return _rawPort.sendPort;
+  }
+
+  StreamSubscription listen(void onData(var message),
+                            { Function onError,
+                              void onDone(),
+                              bool cancelOnError }) {
+    return _controller.stream.listen(onData);
+  }
+
+  close() {
+    _rawPort.close();
+    _controller.close();
+  }
+
+  final RawReceivePort _rawPort;
+  StreamController _controller;
+}
+
+class _RawReceivePortImpl implements RawReceivePort {
+  factory _RawReceivePortImpl() native "RawReceivePortImpl_factory";
+
   close() {
     _portMap.remove(_id);
     _closeInternal(_id);
   }
 
-  SendPort toSendPort() {
+  SendPort get sendPort {
     return new _SendPortImpl(_id);
   }
 
   /**** Internal implementation details ****/
-  // Called from the VM to create a new ReceivePort instance.
-  static _ReceivePortImpl _get_or_create(int id) {
+  // Called from the VM to create a new RawReceivePort instance.
+  static _RawReceivePortImpl _get_or_create(int id) {
     if (_portMap != null) {
-      _ReceivePortImpl port = _portMap[id];
+      _RawReceivePortImpl port = _portMap[id];
       if (port != null) {
         return port;
       }
     }
-    return new _ReceivePortImpl._internal(id);
+    return new _RawReceivePortImpl._internal(id);
   }
 
-  _ReceivePortImpl._internal(int id) : _id = id {
+  _RawReceivePortImpl._internal(int id) : _id = id {
     if (_portMap == null) {
       _portMap = new Map();

[Ivan Posva, 2013/10/25 07:01:46]

This does not need to be a DoublyLinkedHashMap, please change to HashMap.

[floitsch, 2013/10/25 13:11:01]

Done.

     }
     _portMap[id] = this;
   }
 
-  // Called from the VM to retrieve the ReceivePort for a message.
-  static _ReceivePortImpl _lookupReceivePort(int id) {
+  // Called from the VM to retrieve the RawReceivePort for a message.
+  static _RawReceivePortImpl _lookupReceivePort(int id) {
     assert(_portMap != null);

[Ivan Posva, 2013/10/25 07:01:46]

These tests would not be needed if it was a static final Map _portMap = new
HashMap() below.

[floitsch, 2013/10/25 13:11:01]

Done.

     return _portMap[id];
   }
 
   // Called from the VM to dispatch to the handler.
-  static void _handleMessage(_ReceivePortImpl port, int replyId, var message) {
+  static void _handleMessage(_RawReceivePortImpl port, int replyId, var message) {

[Lasse Reichstein Nielsen, 2013/10/25 09:42:49]

long line.

[floitsch, 2013/10/25 13:11:01]

Done.

     assert(port != null);
-    SendPort replyTo = (replyId == 0) ? null : new _SendPortImpl(replyId);
-    (port._onMessage)(message, replyTo);
+    port._handler(message);
   }
 
   // Call into the VM to close the VM maintained mappings.
-  static _closeInternal(int id) native "ReceivePortImpl_closeInternal";
+  static _closeInternal(int id) native "RawReceivePortImpl_closeInternal";
+
+  void set handler(Function newHandler) {
+    this._handler = newHandler;
+  }
 
   final int _id;
-  var _onMessage;
+  Function _handler;
 
-  // id to ReceivePort mapping.
+  // id to RawReceivePort mapping.
   static Map _portMap;
 }
 
 
 class _SendPortImpl implements SendPort {
   /*--- public interface ---*/
   void send(var message, [SendPort replyTo = null]) {

[Ivan Posva, 2013/10/25 07:01:46]

As far as I understand the SendPort.send does not accept a replyTo anymore.

[floitsch, 2013/10/25 13:11:01]

As discussed. We still need it for Native C communication.

     this._sendNow(message, replyTo);
   }
 
   void _sendNow(var message, SendPort replyTo) {

[Ivan Posva, 2013/10/25 07:01:46]

This function makes no sense now. How about calling _sendInternal(_id, 0,
message) from send directly.

0 is an illegal/unmapped port id and then we'll remove the handling for replyTo
ports within the VM once this change has landed.

[floitsch, 2013/10/25 13:11:01]

ditto.

     int replyId = (replyTo == null) ? 0 : replyTo._id;
     _sendInternal(_id, replyId, message);
   }
 
-  Future call(var message) {
-    final completer = new Completer.sync();
-    final port = new _ReceivePortImpl();
-    send(message, port.toSendPort());
-    port.receive((value, ignoreReplyTo) {
-      port.close();
-      if (value is Exception) {
-        completer.completeError(value);
-      } else {
-        completer.complete(value);
-      }
-    });
-    return completer.future;
-  }
-
   bool operator==(var other) {
     return (other is _SendPortImpl) && _id == other._id;
   }
 
   int get hashCode {
     return _id;

[Ivan Posva, 2013/10/25 07:01:46]

We should not leak the id out ever.

[Lasse Reichstein Nielsen, 2013/10/25 09:42:49]

The value needs to deterministically depend on _id only, so we'd need some sort
of hash of the _id instead.
How about a single "Jenkins" step?
    const int MASK = 0x3FFFFFFF; 
    int hash = _id;
    hash = (hash + ((hash & (MASK >> 10)) << 10)) & MASK;
    hash ^= (hash >> 6);
    hash = (hash + ((hash & (MASK >> 3)) << 3)) & MASK;
    hash ^= (hash >> 11);
    hash = (hash + ((hash & (MASK >> 15)) << 15)) & MASK;
    return hash;

[floitsch, 2013/10/25 13:11:01]

Done.

[floitsch, 2013/10/25 13:11:01]

Implemented the single Jenkins hash step as suggested by Lasse.

[Ivan Posva, 2013/10/25 18:41:39]

As long as we cannot go from the hash back to the id, I am fine.

   }
 
   /*--- private implementation ---*/
   const _SendPortImpl(int id) : _id = id;
 
   // _SendPortImpl._create is called from the VM when a new SendPort instance is
   // needed by the VM code.
   static SendPort _create(int id) {
     return new _SendPortImpl(id);
   }
 
   // Forward the implementation of sending messages to the VM. Only port ids
   // are being handed to the VM.
   static _sendInternal(int sendId, int replyId, var message)
       native "SendPortImpl_sendInternal_";
 
   final int _id;
 }
 
 _getPortInternal() native "isolate_getPortInternal";
 
-ReceivePort _portInternal;
+typedef _MainFunction();
+typedef _MainFunctionArgs(args);
+typedef _MainFunctionArgsMessage(args, message);
 
-patch class _Isolate {
-  /* patch */ static ReceivePort get port {
-    if (_portInternal == null) {
-      _portInternal = _getPortInternal();
+/**
+ * Takes the real entry point as argument and invokes it with the initial
+ * message.
+ *
+ * The initial message is (currently) received through the global port variable.
+ */
+void _startIsolate(Function entryPoint, bool isFunctionSpawn) {
+  Isolate._port.first.then((message) {

[Ivan Posva, 2013/10/25 07:01:46]

You cannot use first here. As far as I understand this will close the main port
of the isolate which makes the VM lose its "control" on the isolate.

[floitsch, 2013/10/25 13:11:01]

As discussed: closing the port for now.

+    var reply = message[0];

[Ivan Posva, 2013/10/25 07:01:46]

Please assign the message[0] and message[1] values to name variables to document
what is going on.

[floitsch, 2013/10/25 13:11:01]

Done.

+    // TODO(floitsch): don't send ok-message if we can't find the entry point.
+    reply.send("started");
+    if (isFunctionSpawn) {
+      entryPoint(message[1]);
+    } else if (entryPoint is _MainFunctionArgsMessage) {

[Ivan Posva, 2013/10/25 07:01:46]

This will not mix the if-statement for spawnFunction and spawnUri with the
dispatch based on arguments in the spawnUri case:

} else {
  // In this case we are spawning based on a spawnUri.
  assert(message.length == 3);
  if (entryPoint is _MainFunctionArgsMesssage) {
...

[floitsch, 2013/10/25 13:11:01]

Done.

+      entryPoint(message[1], message[2]);

[Ivan Posva, 2013/10/25 07:01:46]

You are not checking whether the message even has 3 values.

[floitsch, 2013/10/25 13:11:01]

It would throw otherwise.
I have added the assert as suggest.

+    } else if (entryPoint is _MainFunctionArgs) {
+      entryPoint(message[1]);
+    } else {
+      entryPoint();
     }
-    return _portInternal;
+  });
+}
+
+patch class Isolate {
+  /* patch */ static Future<Isolate> spawn(
+      void entryPoint(message), var message) {
+    Completer completer = new Completer<Isolate>.sync();
+    try {
+      // The VM will invoke [_startIsolate] with entryPoint as argument.
+      SendPort controlPort = _spawnFunction(entryPoint);
+      RawReceivePort readyPort = new RawReceivePort();
+      controlPort.send([readyPort.sendPort, message]);
+      readyPort.handler = (_) {

[Ivan Posva, 2013/10/25 07:01:46]

... = (readyMsg) {
  assert(readyMsg == "started");
...

[floitsch, 2013/10/25 13:11:01]

Done.

+        readyPort.close();
+        completer.complete(new Isolate._fromControlPort(controlPort));

[Ivan Posva, 2013/10/25 07:01:46]

See note above. By the time you get here the controlPort has been closed on the
started isolate.

[floitsch, 2013/10/25 13:11:01]

Agreed.
As discussed. Leaving as is. Currently nobody accesses the control port.

+      };
+    } catch(e, st) {
+      // TODO(floitsch): we want errors to go into the returned future.
+      rethrow;
+    };
+    return completer.future;
   }
 
-  /* patch */ static SendPort spawnFunction(void topLevelFunction(),
-      [bool unhandledExceptionCallback(IsolateUnhandledException e)])
+  /* patch */ static Future<Isolate> spawnUri(
+      Uri uri, List<String> args, var message) {
+    Completer completer = new Completer<Isolate>.sync();
+    try {
+      // The VM will invoke [_startIsolate] and not `main`.
+      SendPort controlPort = _spawnUri(uri.path);
+      RawReceivePort readyPort = new RawReceivePort();
+      controlPort.send([readyPort.sendPort, args, message]);
+      readyPort.handler = (_) {

[Ivan Posva, 2013/10/25 07:01:46]

ditto

[floitsch, 2013/10/25 13:11:01]

ditto.

+        readyPort.close();
+        completer.complete(new Isolate._fromControlPort(controlPort));
+      };
+    } catch(e, st) {
+      // TODO(floitsch): we want errors to go into the returned future.
+      rethrow;
+    };
+    return completer.future;
+  }
+
+  static final ReceivePort _port =
+      new ReceivePort.fromRawReceivePort(_getPortInternal());
+
+  static SendPort _spawnFunction(Function topLevelFunction)
       native "isolate_spawnFunction";
 
-  /* patch */ static SendPort spawnUri(String uri) native "isolate_spawnUri";
+  static SendPort _spawnUri(String uri) native "isolate_spawnUri";
 }