OOB messages and general message refactor.

runtime/vm/dart_api_impl.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 "include/dart_api.h"
 
 #include "vm/bigint_operations.h"
 #include "vm/class_finalizer.h"
 #include "vm/compiler.h"
 #include "vm/dart.h"
 #include "vm/dart_api_impl.h"
 #include "vm/dart_api_state.h"
 #include "vm/dart_entry.h"
 #include "vm/debuginfo.h"
 #include "vm/exceptions.h"
 #include "vm/growable_array.h"
 #include "vm/longjump.h"
+#include "vm/message_queue.h"
 #include "vm/native_entry.h"
 #include "vm/object.h"
 #include "vm/object_store.h"
 #include "vm/port.h"
 #include "vm/resolver.h"
 #include "vm/snapshot.h"
 #include "vm/stack_frame.h"
 #include "vm/timer.h"
 #include "vm/verifier.h"
 
@@ skipping 160 matching lines @@
   ASSERT(state.IsValidPersistentHandle(object));
   return reinterpret_cast<PersistentHandle*>(object);
 }
 
 
 Dart_Isolate Api::CastIsolate(Isolate* isolate) {
   return reinterpret_cast<Dart_Isolate>(isolate);
 }
 
 
-Dart_Message Api::CastMessage(uint8_t* message) {
-  return reinterpret_cast<Dart_Message>(message);
-}
-
-
 Dart_Handle Api::Success() {
   Isolate* isolate = Isolate::Current();
   ASSERT(isolate != NULL);
   ApiState* state = isolate->api_state();
   ASSERT(state != NULL);
   PersistentHandle* true_handle = state->True();
   return reinterpret_cast<Dart_Handle>(true_handle);
 }
 
 
@@ skipping 447 matching lines @@
     FATAL1("%s expects argument 'isolate' to be non-null.",  CURRENT_FUNC);
   }
   Isolate* iso = reinterpret_cast<Isolate*>(isolate);
   iso->ScheduleInterrupts(Isolate::kApiInterrupt);
 }
 
 
 // --- Messages and Ports ---
 
 
-DART_EXPORT void Dart_SetMessageCallbacks(
-    Dart_PostMessageCallback post_message_callback,
-    Dart_ClosePortCallback close_port_callback) {
+DART_EXPORT void Dart_SetMessageNotifyCallback(
+    Dart_MessageNotifyCallback message_notify_callback) {
   Isolate* isolate = Isolate::Current();
   CHECK_ISOLATE(isolate);
-  ASSERT(post_message_callback != NULL);
-  ASSERT(close_port_callback != NULL);
-  isolate->set_post_message_callback(post_message_callback);
-  isolate->set_close_port_callback(close_port_callback);
+  isolate->set_message_notify_callback(message_notify_callback);
 }
 
 
 DART_EXPORT Dart_Handle Dart_RunLoop() {
   Isolate* isolate = Isolate::Current();
   DARTSCOPE(isolate);
 
   LongJump* base = isolate->long_jump_base();
   LongJump jump;
   Dart_Handle result;
@@ skipping 20 matching lines @@
   ASSERT(snapshot->IsMessageSnapshot());
 
   // Read object back from the snapshot.
   SnapshotReader reader(snapshot, Isolate::Current());
   Instance& instance = Instance::Handle();
   instance ^= reader.ReadObject();
   return instance.raw();
 }
 
 
-DART_EXPORT Dart_Handle Dart_HandleMessage(Dart_Port dest_port_id,
-                                           Dart_Port reply_port_id,
-                                           Dart_Message dart_message) {
-  DARTSCOPE(Isolate::Current());
-  const Instance& msg = Instance::Handle(DeserializeMessage(dart_message));
-  // TODO(turnidge): Should this call be wrapped in a longjmp?
-  const Object& result =
-      Object::Handle(DartLibraryCalls::HandleMessage(dest_port_id,
-                                                     reply_port_id,
-                                                     msg));
-  if (result.IsError()) {
-    return Api::NewLocalHandle(result);
-  }
-  ASSERT(result.IsNull());
+DART_EXPORT Dart_Handle Dart_HandleMessage() {
+  Isolate* isolate = Isolate::Current();
+  // Process all OOB messages and at most one normal message.
+  Message* message = NULL;
+  Message::Priority priority = Message::kNormalPriority;
+  do {
+    DARTSCOPE(isolate);
+    message = isolate->message_queue()->DequeueNoWait();
+    if (message == NULL) {
+      break;
+    }
+    priority = message->priority();
+    if (priority == Message::kOOBPriority) {
+      UNIMPLEMENTED();

[Ivan Posva, 2012/01/23 22:00:43]

Can you briefly describe in a comment what should happen in this case?

[turnidge, 2012/01/23 22:32:52]

Done.

+    }
+    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.IsError()) {
+      return Api::NewLocalHandle(result);
+    }
+    ASSERT(result.IsNull());
+  } while (priority == Message::kOOBPriority);

[Ivan Posva, 2012/01/23 22:00:43]

How about:
while (priority >= Message::kOOBPriority)

This would break if we ever start using a priority higher than out-of-band.

[turnidge, 2012/01/23 22:32:52]

Anton didn't like the >= because it is unclear which way priorities are ordered,
so I decided to switch this.  But I am now feeling like switching it back.

Done.

On 2012/01/23 22:00:43, Ivan Posva wrote:

   return Api::Success();
 }
 
 
 DART_EXPORT bool Dart_HasLivePorts() {
   Isolate* isolate = Isolate::Current();
   ASSERT(isolate);
   return isolate->live_ports() > 0;
 }
 
 
 static uint8_t* allocator(uint8_t* ptr, intptr_t old_size, intptr_t new_size) {
   void* new_ptr = realloc(reinterpret_cast<void*>(ptr), new_size);
   return reinterpret_cast<uint8_t*>(new_ptr);
 }
 
 
 DART_EXPORT bool Dart_PostIntArray(Dart_Port port_id,
                                    intptr_t len,
                                    intptr_t* data) {
   uint8_t* buffer = NULL;
   MessageWriter writer(&buffer, &allocator);
 
   writer.WriteMessage(len, data);
 
   // Post the message at the given port.
-  return PortMap::PostMessage(port_id, kNoReplyPort, Api::CastMessage(buffer));
+  return PortMap::PostMessage(new Message(
+      port_id, Message::kIllegalPort, buffer, Message::kNormalPriority));
 }
 
 
 DART_EXPORT bool Dart_Post(Dart_Port port_id, Dart_Handle handle) {
   Isolate* isolate = Isolate::Current();
   CHECK_ISOLATE(isolate);
   DARTSCOPE_NOCHECKS(isolate);
   const Object& object = Object::Handle(Api::UnwrapHandle(handle));
   uint8_t* data = NULL;
   SnapshotWriter writer(Snapshot::kMessage, &data, &allocator);
   writer.WriteObject(object.raw());
   writer.FinalizeBuffer();
-  return PortMap::PostMessage(port_id, kNoReplyPort, Api::CastMessage(data));
+  return PortMap::PostMessage(new Message(
+      port_id, Message::kIllegalPort, data, Message::kNormalPriority));
 }
 
 
 DART_EXPORT Dart_Handle Dart_NewSendPort(Dart_Port port_id) {
   Isolate* isolate = Isolate::Current();
   DARTSCOPE(isolate);
   const String& class_name = String::Handle(String::NewSymbol("SendPortImpl"));
   const String& function_name = String::Handle(String::NewSymbol("_create"));
   const int kNumArguments = 1;
   const Array& kNoArgumentNames = Array::Handle();
@@ skipping 1719 matching lines @@
     }
     delete debug_region;
   } else {
     *buffer = NULL;
     *buffer_size = 0;
   }
 }
 
 
 }  // namespace dart