OOB messages and general message refactor.

runtime/vm/isolate.cc

 // 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.
 
 #include "vm/isolate.h"
 
 #include "include/dart_api.h"
 
 #include "vm/assert.h"
 #include "vm/bigint_store.h"
@@ skipping 18 matching lines @@
 namespace dart {
 
 DEFINE_FLAG(bool, report_invocation_count, false,
     "Count function invocations and report.");
 DECLARE_FLAG(bool, generate_gdb_symbols);
 
 
 Isolate::Isolate()
     : store_buffer_(),
       message_queue_(NULL),
-      post_message_callback_(NULL),
-      close_port_callback_(NULL),
+      message_notify_callback_(NULL),
       num_ports_(0),
       live_ports_(0),
       main_port_(0),
       heap_(NULL),
       object_store_(NULL),
       top_resource_(NULL),
       top_context_(Context::null()),
       current_zone_(NULL),
 #if defined(DEBUG)
       no_gc_scope_depth_(0),
@@ skipping 25 matching lines @@
   // Do not delete stack resources: top_resource_ and current_zone_.
   delete bigint_store_;
   delete api_state_;
   delete stub_code_;
   delete code_index_table_;
   delete mutex_;
   mutex_ = NULL;  // Fail fast if interrupts are scheduled on a dead isolate.
 }
 
 
-static bool StandardPostMessageCallback(Dart_Isolate dart_isolate,
-                                        Dart_Port dest_port,
-                                        Dart_Port reply_port,
-                                        Dart_Message dart_message) {
-  Isolate* isolate = reinterpret_cast<Isolate*>(dart_isolate);
-  ASSERT(isolate != NULL);
-  PortMessage* message = new PortMessage(dest_port, reply_port, dart_message);
-  isolate->message_queue()->Enqueue(message);
-  return true;
+void Isolate::PostMessage(Message* message) {
+  message_queue()->Enqueue(message);
+  if (message->priority() <= Message::kOOBPriority) {
+    // Handle out of band messages even if the isolate is busy.
+    ScheduleInterrupts(Isolate::kMessageInterrupt);
+  }
+
+  Dart_MessageNotifyCallback callback = message_notify_callback();
+  if (callback) {
+    // Allow the embedder to handle message notification.
+    (*callback)(Api::CastIsolate(this));
+  }
 }
 
 
-static void StandardClosePortCallback(Dart_Isolate dart_isolate,
-                                      Dart_Port port) {
-  // Remove the pending messages for this port.
-  Isolate* isolate = reinterpret_cast<Isolate*>(dart_isolate);
-  ASSERT(isolate != NULL);
-  if (port == kCloseAllPorts) {
-    isolate->message_queue()->FlushAll();
-  } else {
-    isolate->message_queue()->Flush(port);
-  }
+void Isolate::ClosePort(Dart_Port port) {
+  message_queue()->Flush(port);
+}
+
+
+void Isolate::CloseAllPorts() {
+  message_queue()->FlushAll();
 }
 
 
 Isolate* Isolate::Init() {
   Isolate* result = new Isolate();
   ASSERT(result != NULL);
 
   // TODO(5411455): For now just set the recently created isolate as
   // the current isolate.
   SetCurrent(result);
 
   // Set up the isolate message queue.
   MessageQueue* queue = new MessageQueue();
   ASSERT(queue != NULL);
   result->set_message_queue(queue);
-  result->set_post_message_callback(&StandardPostMessageCallback);
-  result->set_close_port_callback(&StandardClosePortCallback);
 
   // Setup the Dart API state.
   ApiState* state = new ApiState();
   ASSERT(state != NULL);
   result->set_api_state(state);
 
   // Initialize stack top and limit in case we are running the isolate in the
   // main thread.
   // TODO(5411455): Need to figure out how to set the stack limit for the
   // main thread.
@@ skipping 162 matching lines @@
   SnapshotReader reader(snapshot, Isolate::Current());
   Instance& instance = Instance::Handle();
   instance ^= reader.ReadObject();
   return instance.raw();
 }
 
 
 
 RawObject* Isolate::StandardRunLoop() {
   ASSERT(long_jump_base() != NULL);
-  ASSERT(post_message_callback() == &StandardPostMessageCallback);
-  ASSERT(close_port_callback() == &StandardClosePortCallback);
+  ASSERT(message_notify_callback() == NULL);
 
   while (live_ports() > 0) {
     ASSERT(this == Isolate::Current());
     Zone zone(this);
     HandleScope handle_scope(this);
 
-    PortMessage* message = message_queue()->Dequeue(0);
+    Message* message = message_queue()->Dequeue(0);
     if (message != NULL) {
+      if (message->priority() == Message::kOOBPriority) {
+        UNIMPLEMENTED();
+      }
       const Instance& msg =
           Instance::Handle(DeserializeMessage(message->data()));
       const Object& result = Object::Handle(
           DartLibraryCalls::HandleMessage(
               message->dest_port(), message->reply_port(), msg));
       delete message;
-      if (result.IsUnhandledException()) {
+      if (result.IsError()) {
         return result.raw();
       }
+      ASSERT(result.IsNull());
     }
   }
 
   // Indicates success.
   return Object::null();
 }
 
 
 void Isolate::VisitObjectPointers(ObjectPointerVisitor* visitor,
                                   bool validate_frames) {
@@ skipping 41 matching lines @@
 }
 
 
 void Isolate::VisitWeakObjectPointers(ObjectPointerVisitor* visitor) {
   if (api_state() != NULL) {
     api_state()->VisitWeakObjectPointers(visitor);
   }
 }
 
 }  // namespace dart