Use the ThreadPool for all isolates and native ports. Previously,

runtime/vm/message_handler_test.cc

Index: runtime/vm/message_handler_test.cc
===================================================================
--- runtime/vm/message_handler_test.cc	(revision 0)
+++ runtime/vm/message_handler_test.cc	(revision 0)
@@ -0,0 +1,193 @@
+// 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/message_handler.h"
+#include "vm/unit_test.h"
+
+namespace dart {
+
+class MessageHandlerTestPeer {
+ public:
+  explicit MessageHandlerTestPeer(MessageHandler* handler)
+      : handler_(handler) {}
+
+  MessageQueue* queue() { return handler_->queue_; }
+  MessageQueue* oob_queue() { return handler_->oob_queue_; }
+
+ private:
+  MessageHandler* handler_;
+};
+
+
+class TestMessageHandler : public MessageHandler {
+ public:
+  TestMessageHandler()
+      : port_buffer_(strdup("")),
+        notify_count_(0),
+        result_(true) {}
+
+  ~TestMessageHandler() {
+    free(port_buffer_);
+  }
+
+  void MessageNotify(Message::Priority priority) {
+    notify_count_++;
+  }
+
+  bool HandleMessage(Message* message) {
+    // For testing purposes, keep a string with a list of the ports
+    // for all messages we receive.
+    intptr_t len =
+        OS::SNPrint(NULL, 0, "%s %d", port_buffer_, message->dest_port()) + 1;
+    char* buffer = reinterpret_cast<char*>(malloc(len));
+    OS::SNPrint(buffer, len, "%s %d", port_buffer_, message->dest_port());
+    free(port_buffer_);
+    port_buffer_ = buffer;
+    delete message;
+    return result_;
+  }
+
+
+  bool Start() {
+    intptr_t len =
+        OS::SNPrint(NULL, 0, "%s start", port_buffer_) + 1;
+    char* buffer = reinterpret_cast<char*>(malloc(len));
+    OS::SNPrint(buffer, len, "%s start", port_buffer_);
+    free(port_buffer_);
+    port_buffer_ = buffer;
+    return true;
+  }
+
+
+  const char* port_buffer() const { return port_buffer_; }
+  int notify_count() const { return notify_count_; }
+
+  void set_result(bool result) { result_ = result; }
+
+ private:
+  char* port_buffer_;
+  int notify_count_;
+  bool result_;
+};
+
+
+bool TestStartFunction(uword data) {
+  return (reinterpret_cast<TestMessageHandler*>(data))->Start();
+}
+
+
+UNIT_TEST_CASE(MessageHandler_PostMessage) {
+  TestMessageHandler handler;
+  MessageHandlerTestPeer handler_peer(&handler);
+  EXPECT_EQ(0, handler.notify_count());
+
+  // Post a message.
+  Message* message = new Message(0, 0, NULL, Message::kNormalPriority);
+  handler.PostMessage(message);
+
+  // The notify callback is called.
+  EXPECT_EQ(1, handler.notify_count());
+
+  // The message has been added to the correct queue.
+  EXPECT(message == handler_peer.queue()->Dequeue());
+  EXPECT(NULL == handler_peer.oob_queue()->Dequeue());
+  delete message;
+
+  // Post an oob message.
+  message = new Message(0, 0, NULL, Message::kOOBPriority);
+  handler.PostMessage(message);
+
+  // The notify callback is called.
+  EXPECT_EQ(2, handler.notify_count());
+
+  // The message has been added to the correct queue.
+  EXPECT(message == handler_peer.oob_queue()->Dequeue());
+  EXPECT(NULL == handler_peer.queue()->Dequeue());
+  delete message;
+}
+
+
+UNIT_TEST_CASE(MessageHandler_ClosePort) {
+  TestMessageHandler handler;
+  MessageHandlerTestPeer handler_peer(&handler);
+  Message* message1 = new Message(1, 0, NULL, Message::kNormalPriority);
+  handler.PostMessage(message1);
+  Message* message2 = new Message(2, 0, NULL, Message::kNormalPriority);
+  handler.PostMessage(message2);
+
+  handler.ClosePort(1);
+
+  // The message on port 1 is dropped from the queue.
+  EXPECT(message2 == handler_peer.queue()->Dequeue());
+  EXPECT(NULL == handler_peer.queue()->Dequeue());
+  delete message2;
+}
+
+
+UNIT_TEST_CASE(MessageHandler_CloseAllPorts) {
+  TestMessageHandler handler;
+  MessageHandlerTestPeer handler_peer(&handler);
+  Message* message1 = new Message(1, 0, NULL, Message::kNormalPriority);
+  handler.PostMessage(message1);
+  Message* message2 = new Message(2, 0, NULL, Message::kNormalPriority);
+  handler.PostMessage(message2);
+
+  handler.CloseAllPorts();
+
+  // All messages are dropped from the queue.
+  EXPECT(NULL == handler_peer.queue()->Dequeue());
+}
+
+
+UNIT_TEST_CASE(MessageHandler_HandleNextMessage) {
+  TestMessageHandler handler;
+  MessageHandlerTestPeer handler_peer(&handler);
+  Message* message1 = new Message(1, 0, NULL, Message::kNormalPriority);
+  handler.PostMessage(message1);
+  Message* message2 = new Message(2, 0, NULL, Message::kNormalPriority);
+  handler.PostMessage(message2);
+  Message* oob_message1 = new Message(3, 0, NULL, Message::kOOBPriority);
+  handler.PostMessage(oob_message1);
+  Message* oob_message2 = new Message(4, 0, NULL, Message::kOOBPriority);
+  handler.PostMessage(oob_message2);
+
+  // We handle both oob messages and a single normal message.
+  EXPECT(handler.HandleNextMessage());
+  EXPECT_STREQ(" 3 4 1", handler.port_buffer());
+  handler.CloseAllPorts();
+}
+
+
+UNIT_TEST_CASE(MessageHandler_HandleOOBMessages) {
+  TestMessageHandler handler;
+  MessageHandlerTestPeer handler_peer(&handler);
+  Message* message1 = new Message(1, 0, NULL, Message::kNormalPriority);
+  handler.PostMessage(message1);
+  Message* message2 = new Message(2, 0, NULL, Message::kNormalPriority);
+  handler.PostMessage(message2);
+  Message* oob_message1 = new Message(3, 0, NULL, Message::kOOBPriority);
+  handler.PostMessage(oob_message1);
+  Message* oob_message2 = new Message(4, 0, NULL, Message::kOOBPriority);
+  handler.PostMessage(oob_message2);
+
+  // We handle both oob messages but no normal messages.
+  EXPECT(handler.HandleOOBMessages());
+  EXPECT_STREQ(" 3 4", handler.port_buffer());
+  handler.CloseAllPorts();
+}
+
+
+UNIT_TEST_CASE(MessageHandler_RunBlocking) {
+  ThreadPool pool;
+  TestMessageHandler handler;
+  Message* message1 = new Message(1, 0, NULL, Message::kNormalPriority);
+  handler.PostMessage(message1);
+  handler.RunBlocking(&pool,
+                      TestStartFunction,
+                      reinterpret_cast<uword>(&handler));
+  EXPECT_STREQ(" start 1", handler.port_buffer());
+}
+
+
+}  // namespace dart