isolates in frog: playing with API improvements

frog/lib/isolate.dart

 // Copyright (c) 2011, 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.
 
 /**
  * A native object that is shared across isolates. This object is visible to all
  * isolates running on the same worker (either UI or background web worker).
  *
  * This is code that is intended to 'escape' the isolate boundaries in order to
  * implement the semantics of friendly isolates in JavaScript. Without this we
@@ skipping 144 matching lines @@
 
 _deserializeMessage(message) {
   if (_globalState.needSerialization) {
     return new Deserializer().deserialize(message);
   } else {
     // Nothing more to do.
     return message;
   }
 }
 
+/** Wait until all ports in a message are resolved. */
+_waitForPendingPorts(var message, void callback()) {
+  final finder = new PendingSendPortFinder();
+  finder.traverse(message);
+  Futures.wait(finder.ports).then((_) => callback());
+}
+
 /** Default worker. */
 class MainWorker {
   int id = 0;
   void postMessage(msg) native "return \$globalThis.postMessage(msg);";
   void set onmessage(f) native "\$globalThis.onmessage = f;";
   void terminate() {}
 }
 
 /**
  * A web worker. This type is also defined in 'dart:dom', but we define it here
@@ skipping 144 matching lines @@
   Function fn;
   String message;
 
   IsolateEvent(this.isolate, this.fn, this.message);
 
   void process() {
     isolate.eval(fn);
   }
 }
 
-/** Implementation of a send port on top of JavaScript. */
-class SendPortImpl implements SendPort {
+/** Common functionality to all send ports. */
+class BaseSendPort implements SendPort {
+  final int _isolateId;
 
-  const SendPortImpl(this._workerId, this._isolateId, this._receivePortId);
-
-  void send(var message, [SendPort replyTo = null]) {
-    if (replyTo !== null && !(replyTo is SendPortImpl)) {
-      throw "SendPort::send: Illegal replyTo type.";
-    }
-    IsolateNatives._sendMessage(_workerId, _isolateId, _receivePortId,
-        _serializeMessage(message), _serializeMessage(replyTo));
-  }
-
-  // TODO(sigmund): get rid of _sendNow (still used in corelib code)
-  void _sendNow(var message, replyTo) { send(message, replyTo); }
+  BaseSendPort(this._isolateId);
 
   ReceivePortSingleShotImpl call(var message) {
     final result = new ReceivePortSingleShotImpl();
     this.send(message, result.toSendPort());
     return result;
   }
 
-  ReceivePortSingleShotImpl _callNow(var message) {
-    final result = new ReceivePortSingleShotImpl();
-    send(message, result.toSendPort());
-    return result;
+  static void checkReplyTo(SendPort replyTo) {
+    if (replyTo !== null
+        && replyTo is! NativeJsSendPort
+        && replyTo is! WorkerSendPort
+        && replyTo is! BufferingSendPort) {
+      throw new Exception("SendPort.send: Illegal replyTo port type");
+    }
   }
 
-  bool operator==(var other) {
-    return (other is SendPortImpl) &&
+  // TODO(sigmund): replace the current SendPort.call with the following:
+  //Future call(var message) {
+  //   final completer = new Completer();
+  //   final port = new ReceivePort.singleShot();
+  //   send(message, port.toSendPort());
+  //   port.receive((value, ignoreReplyTo) {
+  //     if (value is Exception) {
+  //       completer.completeException(value);
+  //     } else {
+  //       completer.complete(value);
+  //     }
+  //   });
+  //   return completer.future;
+  //}
+
+  abstract void send(var message, [SendPort replyTo]);
+  abstract bool operator ==(var other);
+  abstract int hashCode();
+}
+
+/** A send port that delivers messages in-memory via native JavaScript calls. */
+class NativeJsSendPort extends BaseSendPort implements SendPort {
+  final ReceivePortImpl _receivePort;
+
+  const NativeJsSendPort(this._receivePort, int isolateId) : super(isolateId);
+
+  void send(var message, [SendPort replyTo = null]) {
+    _waitForPendingPorts([message, replyTo], () {
+      checkReplyTo(replyTo);
+      // Check that the isolate still runs and the port is still open
+      final isolate = _globalState.isolates[_isolateId];
+      if (isolate == null) return;
+      if (_receivePort._callback == null) return;
+
+      // messages from WorkerSendPorts get forwarded by [_processWorkerMessage],
+      // in that case we no isolate is currently active, and the message was
+      // already serialized and deserialized.
+      final shouldSerialize = _globalState.currentContext == null;
+      _globalState.topEventLoop.enqueue(isolate, () {
+        if (_receivePort._callback != null) {
+          if (shouldSerialize) {
+            // Force serialization/deserialization as a simple way to ensure
+            // isolate communication restrictions are respected.
+            message = _deserializeMessage(_serializeMessage(message));
+            replyTo = _deserializeMessage(_serializeMessage(replyTo));
+          }
+          _receivePort._callback(message, replyTo);
+        }
+      }, 'receive ' + message);
+    });
+  }
+
+  bool operator ==(var other) => (other is NativeJsSendPort) &&
+      (_receivePort == other._receivePort);
+
+  int hashCode() => _receivePort._id;
+}
+
+/** A send port that delivers messages via worker.postMessage. */
+class WorkerSendPort extends BaseSendPort implements SendPort {
+  final int _workerId;
+  final int _receivePortId;
+
+  const WorkerSendPort(this._workerId, int isolateId, this._receivePortId)
+      : super(isolateId);
+
+  void send(var message, [SendPort replyTo = null]) {
+    _waitForPendingPorts([message, replyTo], () {
+      checkReplyTo(replyTo);
+      final workerMessage = _serializeMessage({
+          'command': 'message',
+          'port': _serializeMessage(this),
+          'msg': message,
+          'replyTo': _serializeMessage(replyTo)});
+
+      if (_globalState.isWorker) {
+        // communication from one worker to another go through the main worker:
+        _globalState.mainWorker.postMessage(workerMessage);
+      } else {
+        _globalState.workers[_workerId].postMessage(workerMessage);
+      }
+    });
+  }
+
+  bool operator ==(var other) {
+    return (other is WorkerSendPort) &&
         (_workerId == other._workerId) &&
         (_isolateId == other._isolateId) &&
         (_receivePortId == other._receivePortId);
   }
 
   int hashCode() {
     return (_workerId << 16) ^ (_isolateId << 8) ^ _receivePortId;
   }
+}
 
-  final int _receivePortId;
-  final int _isolateId;
-  final int _workerId;
+/** A port that buffers messages until an underlying port gets resolve. */
+class BufferingSendPort extends BaseSendPort implements SendPort {
+  static int _bufferingCount = 0;
+
+  /** For implementing equals and hashcode. */
+  final int id;
+
+  /** Underlying port, when resolved. */
+  SendPort _port;
+
+  /**
+   * Future of the underlying port, so that we can detect when this port can be
+   * sent on messages.
+   */
+  Future<SendPort> _futurePort;
+
+  /** Pending messages (and reply ports). */
+  List pending;
+
+  BufferingSendPort(isolateId, this._futurePort)
+      : super(isolateId), id = _bufferingCount, pending = [] {
+    _bufferingCount++;
+    _futurePort.then((p) {
+      _port = p;
+      for (final message in pending) {
+        p.send(message[0], message[1]);
+      }
+      pending = null;
+    });
+  }
+
+  BufferingSendPort.fromPort(isolateId, this._port)
+      : super(isolateId), id = _bufferingCount {
+    _bufferingCount++;
+  }
+
+  void send(var message, [SendPort replyTo]) {
+    if (_port != null) {
+      _port.send(message, replyTo);
+    } else {
+      pending.add([message, replyTo]);
+    }
+  }
+
+  bool operator ==(var other) => (other is BufferingSendPort && id == other.id);
+  int hashCode() => id;
 }
 
 /** Default factory for receive ports. */
 class ReceivePortFactory {
 
   factory ReceivePort() {
     return new ReceivePortImpl();
   }
 
   factory ReceivePort.singleShot() {
     return new ReceivePortSingleShotImpl();
   }
 }
 
 /** Implementation of a multi-use [ReceivePort] on top of JavaScript. */
 class ReceivePortImpl implements ReceivePort {
+  int _id;
+  Function _callback;
+  static int _nextFreeId = 1;
+
   ReceivePortImpl()
       : _id = _nextFreeId++ {
     _globalState.currentContext.register(_id, this);
   }
 
   void receive(void onMessage(var message, SendPort replyTo)) {
     _callback = onMessage;
   }
 
   void close() {
     _callback = null;
     _globalState.currentContext.unregister(_id);
   }
 
-  /**
-   * Returns a fresh [SendPort]. The implementation is not allowed to cache
-   * existing ports.
-   */
   SendPort toSendPort() {
-    return new SendPortImpl(
-        _globalState.currentWorkerId, _globalState.currentContext.id, _id);
+    return new NativeJsSendPort(this, _globalState.currentContext.id);
   }
-
-  int _id;
-  Function _callback;
-
-  static int _nextFreeId = 1;
 }
 
 /** Implementation of a single-shot [ReceivePort]. */
 class ReceivePortSingleShotImpl implements ReceivePort {
 
   ReceivePortSingleShotImpl() : _port = new ReceivePortImpl() { }
 
   void receive(void callback(var message, SendPort replyTo)) {
     _port.receive((var message, SendPort replyTo) {
       _port.close();
@@ skipping 33 matching lines @@
 
     return completer.future;
   }
 
   static SendPort _startWorker(Isolate runnable, SendPort replyPort) {
     var factoryName = _getJSConstructorName(runnable);
     if (_globalState.isWorker) {
       _globalState.mainWorker.postMessage(_serializeMessage({
           'command': 'spawn-worker',
           'factoryName': factoryName,
-          'replyPort': replyPort}));
+          'replyPort': _serializeMessage(replyPort)}));
     } else {
       _spawnWorker(factoryName, _serializeMessage(replyPort));
     }
   }
 
-
   /**
    * The src url for the script tag that loaded this code. Used to create
    * JavaScript workers.
    */
   static String get _thisScript() =>
       _thisScriptCache != null ? _thisScriptCache : _computeThisScript();
 
   static String _thisScriptCache;
 
   // TODO(sigmund): fix - this code should be run synchronously when loading the
@@ skipping 45 matching lines @@
    */
   static _getEventData(e) native "return e.data";
 
   /**
    * Process messages on a worker, either to control the worker instance or to
    * pass messages along to the isolate running in the worker.
    */
   static void _processWorkerMessage(sender, e) {
     var msg = _deserializeMessage(_getEventData(e));
     switch (msg['command']) {
+      // TODO(sigmund): delete after we migrate to Isolate2
       case 'start':
         _globalState.currentWorkerId = msg['id'];
         var runnerObject =
             _allocate(_getJSConstructorFromName(msg['factoryName']));
         var serializedReplyTo = msg['replyTo'];
         _globalState.topEventLoop.enqueue(new IsolateContext(), function() {
           var replyTo = _deserializeMessage(serializedReplyTo);
-          IsolateNatives._startIsolate(runnerObject, replyTo);
+          _startIsolate(runnerObject, replyTo);
         }, 'worker-start');
         _globalState.topEventLoop.run();
         break;
+      case 'start2':
+        _globalState.currentWorkerId = msg['id'];
+        Function entryPoint = _getJSFunctionFromName(msg['functionName']);
+        var replyTo = _deserializeMessage(msg['replyTo']);
+        _globalState.topEventLoop.enqueue(new IsolateContext(), function() {
+          _startIsolate2(entryPoint, replyTo);
+        }, 'worker-start');
+        _globalState.topEventLoop.run();
+        break;
+      // TODO(sigmund): delete after we migrate to Isolate2
       case 'spawn-worker':
         _spawnWorker(msg['factoryName'], msg['replyPort']);
         break;
+      case 'spawn-worker2':
+        _spawnWorker2(msg['functionName'], msg['uri'], msg['replyPort']);
+        break;
       case 'message':
-        _sendMessage(msg['workerId'], msg['isolateId'], msg['portId'],
-            msg['msg'], msg['replyTo']);
+        final iid = _globalState.currentContext == null ? '?' : '${_globalState.currentContext.id }';
+        final port = _deserializeMessage(msg['port']);
+        port.send(msg['msg'], _deserializeMessage(msg['replyTo']));
         _globalState.topEventLoop.run();
         break;
       case 'close':
         _log("Closing Worker");
         _globalState.workers.remove(sender.id);
         sender.terminate();
         _globalState.topEventLoop.run();
         break;
       case 'log':
         _log(msg['msg']);
@@ skipping 40 matching lines @@
   // TODO(sigmund): find a browser-generic way to support this.
   static var _getJSConstructorName(Isolate runnable) native """
     return runnable.constructor.name;
   """;
 
   /** Find a constructor given it's name. */
   static var _getJSConstructorFromName(String factoryName) native """
     return \$globalThis[factoryName];
   """;
 
+  static var _getJSFunctionFromName(String functionName) native """
+    return \$globalThis[functionName];
+  """;
+
+  static String _getJSFunctionName(Function f) native "return f.name || null;";
+
   /** Create a new JavasSript object instance given it's constructor. */
   static var _allocate(var ctor) native "return new ctor();";
 
   /** Starts a non-worker isolate. */
   static SendPort _startNonWorker(Isolate runnable, SendPort replyTo) {
     // Spawn a new isolate and create the receive port in it.
     final spawned = new IsolateContext();
 
     // Instead of just running the provided runnable, we create a
     // new cloned instance of it with a fresh state in the spawned
     // isolate. This way, we do not get cross-isolate references
     // through the runnable.
     final ctor = _getJSConstructor(runnable);
     _globalState.topEventLoop.enqueue(spawned, function() {
       _startIsolate(_allocate(ctor), replyTo);
     }, 'nonworker start');
   }
 
   /** Given a ready-to-start runnable, start running it. */
   static void _startIsolate(Isolate isolate, SendPort replyTo) {
     _fillStatics(_globalState.currentContext);
     ReceivePort port = new ReceivePort();
     replyTo.send(_SPAWNED_SIGNAL, port.toSendPort());
     isolate._run(port);
   }
 
-  static void _sendMessage(int workerId, int isolateId, int receivePortId,
-      message, replyTo) {
-    // Both the message and the replyTo are already serialized.
-    if (workerId == _globalState.currentWorkerId) {
-      var isolate = _globalState.isolates[isolateId];
-      if (isolate == null) return;  // Isolate has been closed.
-      var receivePort = isolate.lookup(receivePortId);
-      if (receivePort == null) return;  // ReceivePort has been closed.
-      _globalState.topEventLoop.enqueue(isolate, () {
-        if (receivePort._callback != null) {
-          receivePort._callback(
-            _deserializeMessage(message), _deserializeMessage(replyTo));
-        }
-      }, 'receive ' + message);
+  // TODO(sigmund): clean up above, after we make the new API the default:
+
+  static _spawn2(String functionName, String uri, bool isLight) {
+    Completer<SendPort> completer = new Completer<SendPort>();
+    ReceivePort port = new ReceivePort.singleShot();
+    port.receive((msg, SendPort replyPort) {
+      assert(msg == _SPAWNED_SIGNAL);
+      completer.complete(replyPort);
+    });
+
+    SendPort signalReply = port.toSendPort();
+
+    if (_globalState.useWorkers && !isLight) {
+      _startWorker2(functionName, uri, signalReply);
     } else {
-      var worker;
-      // communication between workers go through the main worker
-      if (_globalState.isWorker) {
-        worker = _globalState.mainWorker;
-      } else {
-        // TODO(sigmund): make sure this works
-        worker = _globalState.workers[workerId];
-      }
-      worker.postMessage(_serializeMessage({
-          'command': 'message',
-          'workerId': workerId,
-          'isolateId': isolateId,
-          'portId': receivePortId,
-          'msg': message,
-          'replyTo': replyTo }));
+      _startNonWorker2(functionName, uri, signalReply);
+    }
+    return new BufferingSendPort(
+        _globalState.currentContext.id, completer.future);
+  }
+
+  static SendPort _startWorker2(
+      String functionName, String uri, SendPort replyPort) {
+    if (_globalState.isWorker) {
+      _globalState.mainWorker.postMessage(_serializeMessage({
+          'command': 'spawn-worker2',
+          'functionName': functionName,
+          'uri': uri,
+          'replyPort': replyPort}));
+    } else {
+      _spawnWorker2(functionName, uri, replyPort);
     }
   }
+
+  static SendPort _startNonWorker2(
+      String functionName, String uri, SendPort replyPort) {
+    _globalState.topEventLoop.enqueue(new IsolateContext(), function() {
+      final func = _getJSFunctionFromName(functionName);
+      _startIsolate2(func, replyPort);
+    }, 'nonworker start');
+  }
+
+  static void _startIsolate2(Function topLevel, SendPort replyTo) {
+    _fillStatics(_globalState.currentContext);
+    final port = new ReceivePort();
+    replyTo.send(_SPAWNED_SIGNAL, port.toSendPort());
+    topLevel(port);
+  }
+
+  /**
+   * Spawns an isolate in a worker. [factoryName] is the Javascript constructor
+   * name for the isolate entry point class.
+   */
+  static void _spawnWorker2(functionName, uri, replyPort) {
+    if (uri == null) uri = _thisScript;
+    final worker = _newWorker(uri);
+    worker.onmessage = (e) { _processWorkerMessage(worker, e); };
+    var workerId = _globalState.nextWorkerId++;
+    // We also store the id on the worker itself so that we can unregister it.
+    worker.id = workerId;
+    _globalState.workers[workerId] = worker;
+    worker.postMessage(_serializeMessage({
+      'command': 'start2',
+      'id': workerId,
+      // Note: we serialize replyPort twice because the child worker needs to
+      // first deserialize the worker id, before it can correctly deserialize
+      // the port (port deserialization is sensitive to what is the current
+      // workerId).
+      'replyTo': _serializeMessage(replyPort),
+      'functionName': functionName }));
+  }
 }
+
+class Isolate2Impl implements Isolate2 {
+  SendPort sendPort;
+
+  Isolate2Impl(this.sendPort);
+
+  void stop() {}
+}
+
+class IsolateFactory implements Isolate2 {
+
+  factory Isolate2.fromCode(Function topLevelFunction) {
+    final name = IsolateNatives._getJSFunctionName(topLevelFunction);
+    if (name == null) {
+      throw new UnsupportedOperationException(
+          "only top-level functions can be spawned.");
+    }
+    return new Isolate2Impl(IsolateNatives._spawn2(name, null, false));
+  }
+
+  factory Isolate2.fromUri(String uri) {
+    return new Isolate2Impl(IsolateNatives._spawn2(null, uri, false));
+  }
+}