Add Isolate.kill().

sdk/lib/_internal/lib/isolate_helper.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.
 
 library _isolate_helper;
 
 import 'dart:async';
 import 'dart:collection' show Queue, HashMap;
 import 'dart:isolate';
 import 'dart:_js_helper' show
@@ skipping 262 matching lines @@
 
   // TODO(lrn): Store these in single "PauseState" object, so they don't take
   // up as much room when not pausing.
   bool isPaused = false;
   List<_IsolateEvent> delayedEvents = [];
   Set<Capability> pauseTokens = new Set();
 
   // Container with the "on exit" handler send-ports.
   var doneHandlers;
 
+  /**
+   * Queue of functions to call when the current event is complete.
+   *
+   * These events are not just put at the front of the event queue, because
+   * they represent control messages, and should be handled even if the
+   * event queue is paused.
+   */
+  var _scheduledControlEvents;
+
   /** Whether errors are considered fatal. */
   // This doesn't do anything yet. We need to be able to catch uncaught errors
   // (oxymoronically) in order to take lethal action. This is waiting for the
   // same change as the uncaught error listeners.
   bool errorsAreFatal = false;
 
   _IsolateContext() {
     this.registerWeak(controlPort._id, controlPort);
   }
 
@@ skipping 32 matching lines @@
     if (doneHandlers == null) return;
     doneHandlers.remove(responsePort);
   }
 
   void setErrorsFatal(Capability authentification, bool errorsAreFatal) {
     if (terminateCapability != authentification) return;
     this.errorsAreFatal = errorsAreFatal;
   }
 
   void handlePing(SendPort responsePort, int pingType) {
-    if (pingType == Isolate.PING_EVENT) {
-      _globalState.topEventLoop.enqueue(this, () {
-        responsePort.send(null);
-      }, "ping");
-    } else {
-      // There is no difference between PING_ALIVE and PING_CONTROL
-      // since we don't handle it before the control event queue.
+    if (pingType == Isolate.OUT_OF_BAND ||
+        (pingType == Isolate.BEFORE_NEXT_EVENT &&
+         !identical(_globalState.currentContext, this))) {

[floitsch, 2014/03/14 13:01:44]

Create a helper function, so that it's more obvious what this is doing.

       responsePort.send(null);
+      return;
     }
+    void respond() { responsePort.send(null); }
+    if (pingType == Isolate.AS_EVENT) {
+      _globalState.topEventLoop.enqueue(this, respond, "ping");
+      return;
+    }
+    assert(pingType == Isolate.BEFORE_NEXT_EVENT);
+    if (_scheduledControlEvents == null) {
+      _scheduledControlEvents = new Queue();
+    }
+    _scheduledControlEvents.addLast(respond);
+  }
+
+  void handleKill(Capability authentification, int priority) {
+    if (this.terminateCapability != authentification) return;
+    if (priority == Isolate.OUT_OF_BAND ||
+        (priority == Isolate.BEFORE_NEXT_EVENT &&
+         !identical(_globalState.currentContext, this))) {
+      kill();
+      return;
+    }
+    if (priority == Isolate.AS_EVENT) {
+      _globalState.topEventLoop.enqueue(this, kill, "kill");
+      return;
+    }
+    assert(pingType == Isolate.BEFORE_NEXT_EVENT);

[floitsch, 2014/03/14 13:01:44]

priority.

[Lasse Reichstein Nielsen, 2014/03/19 13:07:51]

Ack.

+    if (_scheduledControlEvents == null) {
+      _scheduledControlEvents = new Queue();
+    }
+    _scheduledControlEvents.addLast(kill);
   }
 
   /**
    * Run [code] in the context of the isolate represented by [this].
    */
   dynamic eval(Function code) {
     var old = _globalState.currentContext;
     _globalState.currentContext = this;
     this._setGlobals();
     var result = null;
     try {
       result = code();
     } finally {
       _globalState.currentContext = old;
       if (old != null) old._setGlobals();
+      if (_scheduledControlEvents != null) {
+        while (_scheduledControlEvents.isNotEmpty) {
+          (_scheduledControlEvents.removeFirst())();
+        }
+      }
     }
     return result;
   }
 
   void _setGlobals() {
     JS_SET_CURRENT_ISOLATE(isolateStatics);
   }
 
+  /**
+   * Handle messages comming in on the control port.
+   *
+   * These events do not go through the event queue.
+   * The `_globalState.currentContext` context is not set to this context
+   * during the handling.
+   */
   void handleControlMessage(message) {
     switch (message[0]) {
       case "pause":
         addPause(message[1], message[2]);
         break;
       case "resume":
         removePause(message[1]);
         break;
       case 'add-ondone':
         addDoneListener(message[1]);
         break;
       case 'remove-ondone':
         removeDoneListener(message[1]);
         break;
       case 'set-errors-fatal':
         setErrorsFatal(message[1], message[2]);
         break;
       case "ping":
         handlePing(message[1], message[2]);
         break;
+      case "kill":
+        handleKill(message[1], message[2]);
+        break;
       default:
-        print("UNKNOWN MESSAGE: $message");
     }
   }
 
   /** Looks up a port registered for this isolate. */
   RawReceivePortImpl lookup(int portId) => ports[portId];
 
   void _addRegistration(int portId, RawReceivePortImpl port) {
     if (ports.containsKey(portId)) {
       throw new Exception("Registry: ports must be registered only once.");
     }
@@ skipping 13 matching lines @@
    */
   void registerWeak(int portId, RawReceivePortImpl port)  {
     weakPorts.add(portId);
     _addRegistration(portId, port);
   }
 
   void _updateGlobalState() {
     if (ports.length - weakPorts.length > 0 || isPaused) {
       _globalState.isolates[id] = this; // indicate this isolate is active
     } else {
-      _shutdown();
+      kill();
     }
   }
 
-  void _shutdown() {
+  void kill() {
+    if (_scheduledControlEvents != null) {
+      // Kill all pending events.
+      _scheduledControlEvents.clear();
+    }
+    // Stop listening on all ports.
+    // This should happen before sending events to done handlers, in case
+    // we are listening on ourselves.
+    // Closes all ports, including control port.
+    for (var port in ports.values) {
+      port._close();
+    }
+    ports.clear();
+    weakPorts.clear();
     _globalState.isolates.remove(id); // indicate this isolate is not active
-    // Send "done" event to all listeners. This must be done after deactivating
-    // the current isolate, or it may get events if listening to itself.
     if (doneHandlers != null) {
       for (SendPort port in doneHandlers) {
         port.send(null);
       }
+      doneHandlers = null;
     }
   }
 
   /** Unregister a port on this isolate. */
   void unregister(int portId) {
     ports.remove(portId);
     weakPorts.remove(portId);
     _updateGlobalState();
   }
 }
@@ skipping 595 matching lines @@
 
   // Creates the control port of an isolate.
   // This is created before the isolate context object itself,
   // so it cannot access the static _nextFreeId field.
   RawReceivePortImpl._controlPort() : _handler = null, _id = 0;
 
   void set handler(Function newHandler) {
     _handler = newHandler;
   }
 
+  // Close the port without unregistering it.
+  // Used by an isolate context to close all ports when shutting down.
+  void _close() {
+    _isClosed = true;
+    _handler = null;
+  }
+
   void close() {
     if (_isClosed) return;
     _isClosed = true;
     _handler = null;
     _globalState.currentContext.unregister(_id);
   }
 
   void _add(dataEvent) {
     if (_isClosed) return;
     _handler(dataEvent);
@@ skipping 529 matching lines @@
   }
 
   bool operator==(Object other) {
     if (identical(other, this)) return true;
     if (other is CapabilityImpl) {
       return identical(_id, other._id);
     }
     return false;
   }
 }