OOB messages and general message refactor.

runtime/vm/isolate.cc

 // 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.
 
 #include "vm/isolate.h"
 
 #include "include/dart_api.h"
 #include "platform/assert.h"
 #include "vm/bigint_store.h"
 #include "vm/code_index_table.h"
@@ skipping 19 matching lines @@
 DEFINE_FLAG(bool, report_invocation_count, false,
             "Count function invocations and report.");
 DEFINE_FLAG(bool, trace_isolates, false,
             "Trace isolate creation and shut down.");
 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),
       name_(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)
@@ skipping 28 matching lines @@
   delete bigint_store_;
   delete api_state_;
   delete stub_code_;
   delete code_index_table_;
   delete debugger_;
   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) {
+  if (FLAG_trace_isolates) {
+    const char* source_name = "<native code>";
+    Isolate* source_isolate = Isolate::Current();
+    if (source_isolate) {
+      source_name = source_isolate->name();
+    }
+    OS::Print("[>] Posting message:\n"
+              "\tsource:     %s\n"
+              "\treply_port: %lld\n"
+              "\tdest:       %s\n"
+              "\tdest_port:  %lld\n",
+              source_name, message->reply_port(), name(), message->dest_port());
+  }
+  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(const char* name_prefix) {
   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 192 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) {
+        // TODO(turnidge): Out of band messages will not go through the
+        // regular message handler.  Instead they will be dispatched to
+        // special vm code.  Implement.
+        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 34 matching lines @@
 }
 
 
 void Isolate::VisitWeakPersistentHandles(HandleVisitor* visitor) {
   if (api_state() != NULL) {
     api_state()->VisitWeakHandles(visitor);
   }
 }
 
 }  // namespace dart