OOB messages and general message refactor.

runtime/vm/message_queue.h

Index: runtime/vm/message_queue.h
===================================================================
--- runtime/vm/message_queue.h	(revision 3557)
+++ runtime/vm/message_queue.h	(working copy)
@@ -10,60 +10,84 @@
 
 namespace dart {
 
-class PortMessage {
+class Message {
  public:
+  typedef enum {
+    kNormalPriority = 0,  // Deliver message when idle.
+    kOOBPriority = 1,     // Deliver message asap.
+
+    // Iteration.
+    kFirstPriority = 0,
+    kNumPriorities = 2,
+  } Priority;
+
+  // A port number which is never used.
+  static const int kIllegalPort = 0;
+
   // A new message to be sent between two isolates. The data handed to this
   // message will be disposed by calling free() once the message object is
   // being destructed (after delivery or when the receiving port is closed).
-  PortMessage(Dart_Port dest_port, Dart_Port reply_port, Dart_Message data)
+  //
+  // If reply_port is kIllegalPort, then there is no reply port.
+  Message(Dart_Port dest_port, Dart_Port reply_port,
+          uint8_t* data, Priority priority)
       : next_(NULL),
         dest_port_(dest_port),
         reply_port_(reply_port),
-        data_(data) {}
-  ~PortMessage() {
+        data_(data),
+        priority_(priority) {}
+  ~Message() {
     free(data_);
   }
 
   Dart_Port dest_port() const { return dest_port_; }
   Dart_Port reply_port() const { return reply_port_; }
-  Dart_Message data() const { return data_; }
+  uint8_t* data() const { return data_; }
+  Priority priority() const { return priority_; }
 
  private:
   friend class MessageQueue;
 
-  PortMessage* next_;
+  Message* next_;
   Dart_Port dest_port_;
   Dart_Port reply_port_;
-  Dart_Message data_;
+  uint8_t* data_;
+  Priority priority_;
 
-  DISALLOW_COPY_AND_ASSIGN(PortMessage);
+  DISALLOW_COPY_AND_ASSIGN(Message);
 };
 
 
 // There is a message queue per isolate.
 class MessageQueue {
  public:
-  MessageQueue() : head_(NULL), tail_(NULL) {}
+  MessageQueue();
   ~MessageQueue();
 
-  void Enqueue(PortMessage* msg);
+  void Enqueue(Message* msg);
 
   // Gets the next message from the message queue, possibly blocking
   // if no message is available. 'millis' is a timeout in
   // milliseconds. If 'millis' is 0, then this means to block
   // indefinitely. May block if no message is available. May return
   // NULL even if 'millis' is 0 due to spurious wakeups.
-  PortMessage* Dequeue(int64_t millis);
+  Message* Dequeue(int64_t millis);
 
+  // Gets the next message from the message queue if available.  Will
+  // not block.
+  Message* DequeueNoWait();
+
   void Flush(Dart_Port port);
   void FlushAll();
 
  private:
   friend class MessageQueueTestPeer;
 
+  Message* DequeueNoWaitHoldsLock();
+
   Monitor monitor_;
-  PortMessage* head_;
-  PortMessage* tail_;
+  Message* head_[Message::kNumPriorities];
+  Message* tail_[Message::kNumPriorities];
 
   DISALLOW_COPY_AND_ASSIGN(MessageQueue);
 };