isolate lib: small refactor to distinguish protocols at the port level

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 308 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 {
+  /** Id for the destination isolate. */
+  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) {
+      throw "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]) {
+    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;
+
+    // We force serialization/deserialization as a simple way to ensure isolate
+    // communication restrictions are respected between isolates that live in
+    // the same worker. NativeJsSendPort delivers both messages from the same
+    // worker and messages from other workers. In particular, messages sent from
+    // a worker via a WorkerSendPort are received at [_processWorkerMessage] and
+    // forwarded to a native port. In such cases, here we'll see
+    // [_globalState.currentContext == null].
+    final shouldSerialize = _globalState.currentContext != null
+        && _globalState.currentContext.id != _isolateId;
+    if (shouldSerialize) {
+      message = _serializeMessage(message);
+      replyTo = _serializeMessage(replyTo);
+    }
+    _globalState.topEventLoop.enqueue(isolate, () {
+      if (_receivePort._callback != null) {
+        if (shouldSerialize) {

[eub, 2012/02/10 21:20:57]

I liked your previous version with the serialize-deserialize in one shot instead
of two if() checks.

[Siggi Cherem (dart-lang), 2012/02/10 22:09:22]

Me too - when I did some testing, I realized that it didn't provide me the right
semantics. If I forced isolates to run in the same place, this example was
printing "after" instead of "before":

other(port) => port.receive((msg, _) => print(msg[0]));

main() {
  sample = ["before"];
  (new Isolate.fromCode(other)).sendPort.send(sample);
  sample[0] = "after";
}

Basically because the serialize-deserialize was applied after sample was
modified.

[eub, 2012/02/10 22:11:50]

Can we come up with a simple comment explaining to my future self why I
shouldn't change the code to the other version? :)

+          message = _deserializeMessage(message);
+          replyTo = _deserializeMessage(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]) {
+    checkReplyTo(replyTo);
+    final workerMessage = _serializeMessage({
+        'command': 'message',
+        'port': this,
+        'msg': message,
+        'replyTo': 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() {
+    // TODO(sigmund): use a standard hash when we get one available in corelib.
     return (_workerId << 16) ^ (_isolateId << 8) ^ _receivePortId;
   }
-
-  final int _receivePortId;
-  final int _isolateId;
-  final int _workerId;
 }
 
 /** 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 120 matching lines @@
         _globalState.topEventLoop.enqueue(new IsolateContext(), function() {
           var replyTo = _deserializeMessage(serializedReplyTo);
           IsolateNatives._startIsolate(runnerObject, replyTo);
         }, 'worker-start');
         _globalState.topEventLoop.run();
         break;
       case 'spawn-worker':
         _spawnWorker(msg['factoryName'], msg['replyPort']);
         break;
       case 'message':
-        _sendMessage(msg['workerId'], msg['isolateId'], msg['portId'],
-            msg['msg'], msg['replyTo']);
+        msg['port'].send(msg['msg'], 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 65 matching lines @@
     }, '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);
-    } 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 }));
-    }
-  }
 }