dart-lang/sdk
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

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

Browse Source 
each isolate or native port had a dedicated thread.

Refactored the MessageHandler api...

- Added a Run function to allow a MessageHandler to run on a
  ThreadPool.  These functions take a start and end callback to allow for
  isolate initialization and shutdown.

- Made the queue private to the MessageHandler and moved all message
  processing code inside the MessageHandler (got rid of all of the
  different flavors of RunLoop).  This helps remove some code
  duplication and hides the details of how messages are handled.

- Moved all locking and notification out of MessageQueue and moved it
  up to MessageHandler.  Moved OOB support out of MessageQueue and up
  to MessageHandler.  These changes make the MessageQueue much
  simpler.

- Refactored native port and isolate MessageHandlers to share more code.

- Improved --trace_isolates output.

- Added tests for MessageHandler.

Refactored lib/isolate code...

- Use the new MessageHandler::Run api.

- Got rid of the LongJump stuff in RunIsolate.  No longer needed.

- Use the new StartIsolateScope/SwitchIsolateScope to make the code
  less verbose and less error-prone.

- Store top-level isolate errors in the sticky_error.

Added StartIsolateScope/SwitchIsolateScope classes.
Review URL: https://chromiumcodereview.appspot.com//9924015

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@6762 260f80e4-7a28-3924-810f-c04153c831b5
This commit is contained in:
turnidge@google.com
2012-04-19 19:47:27 +00:00
parent 8736ce1aaf
commit 9d72035ce5
19 changed files with 1006 additions and 762 deletions
Download Patch File Download Diff File Expand all files Collapse all files
runtime/lib/isolate.cc
+63 -78
View File
@@ -10,7 +10,7 @@
#include "vm/dart_entry.h"
#include "vm/exceptions.h"
#include "vm/longjump.h"
#include "vm/message.h"
#include "vm/message_handler.h"
#include "vm/object.h"
#include "vm/object_store.h"
#include "vm/port.h"
@@ -22,16 +22,13 @@ namespace dart {
class IsolateStartData {
public:
IsolateStartData(Isolate* isolate,
char* library_url,
IsolateStartData(char* library_url,
char* class_name,
intptr_t port_id)
: isolate_(isolate),
library_url_(library_url),
: library_url_(library_url),
class_name_(class_name),
port_id_(port_id) {}
Isolate* isolate_;
char* library_url_;
char* class_name_;
intptr_t port_id_;
@@ -53,14 +50,11 @@ static uint8_t* SerializeObject(const Instance& obj) {
}
// TODO(turnidge): Taking down the whole vm when an isolate fails is
// bad. Change this.
static void ProcessError(const Object& obj) {
static void StoreError(Isolate* isolate, const Object& obj) {
ASSERT(obj.IsError());
Error& error = Error::Handle();
error ^= obj.raw();
OS::PrintErr("%s\n", error.ToErrorCString());
exit(255);
isolate->object_store()->set_sticky_error(error);
}
@@ -113,31 +107,29 @@ RawObject* ReceivePortCreate(intptr_t port_id) {
}
static void RunIsolate(uword parameter) {
IsolateStartData* data = reinterpret_cast<IsolateStartData*>(parameter);
Isolate* isolate = data->isolate_;
static bool RunIsolate(uword parameter) {
Isolate* isolate = reinterpret_cast<Isolate*>(parameter);
IsolateStartData* data =
reinterpret_cast<IsolateStartData*>(isolate->spawn_data());
isolate->set_spawn_data(NULL);
char* library_url = data->library_url_;
char* class_name = data->class_name_;
intptr_t port_id = data->port_id_;
delete data;
Isolate::SetCurrent(isolate);
// Intialize stack limit in case we are running isolate in a
// different thread than in which it was initialized.
isolate->SetStackLimitFromCurrentTOS(reinterpret_cast<uword>(&isolate));
LongJump* base = isolate->long_jump_base();
LongJump jump;
isolate->set_long_jump_base(&jump);
if (setjmp(*jump.Set()) == 0) {
{
StartIsolateScope start_scope(isolate);
Zone zone(isolate);
HandleScope handle_scope(isolate);
ASSERT(ClassFinalizer::FinalizePendingClasses());
// Lookup the target class by name, create an instance and call the run
// method.
const String& lib_name = String::Handle(String::NewSymbol(library_url));
free(library_url);
const Library& lib = Library::Handle(Library::LookupLibrary(lib_name));
ASSERT(!lib.IsNull());
const String& cls_name = String::Handle(String::NewSymbol(class_name));
free(class_name);
const Class& target_class = Class::Handle(lib.LookupClass(cls_name));
// TODO(iposva): Deserialize or call the constructor after allocating.
// For now, we only support a non-parameterized or raw target class.
@@ -158,7 +150,8 @@ static void RunIsolate(uword parameter) {
arguments,
kNoArgumentNames);
if (result.IsError()) {
ProcessError(result);
StoreError(isolate, result);
return false;
}
ASSERT(result.IsNull());
}
@@ -171,7 +164,8 @@ static void RunIsolate(uword parameter) {
// TODO(iposva): Allocate the proper port number here.
const Object& local_port = Object::Handle(ReceivePortCreate(port_id));
if (local_port.IsError()) {
ProcessError(local_port);
StoreError(isolate, local_port);
return false;
}
GrowableArray<const Object*> arguments(1);
arguments.Add(&local_port);
@@ -181,28 +175,35 @@ static void RunIsolate(uword parameter) {
arguments,
kNoArgumentNames);
if (result.IsError()) {
ProcessError(result);
}
ASSERT(result.IsNull());
free(class_name);
free(library_url);
result = isolate->StandardRunLoop();
if (result.IsError()) {
ProcessError(result);
StoreError(isolate, result);
return false;
}
ASSERT(result.IsNull());
}
return true;
}
} else {
static void ShutdownIsolate(uword parameter) {
Isolate* isolate = reinterpret_cast<Isolate*>(parameter);
{
// Print the error if there is one. This may execute dart code to
// print the exception object, so we need to use a StartIsolateScope.
StartIsolateScope start_scope(isolate);
Zone zone(isolate);
HandleScope handle_scope(isolate);
const Error& error = Error::Handle(
Isolate::Current()->object_store()->sticky_error());
const char* errmsg = error.ToErrorCString();
OS::PrintErr("%s\n", errmsg);
exit(255);
Error& error = Error::Handle();
error = isolate->object_store()->sticky_error();
if (!error.IsNull()) {
OS::PrintErr("%s\n", error.ToErrorCString());
exit(255);
}
}
{
// Shut the isolate down.
SwitchIsolateScope switch_scope(isolate);
Dart::ShutdownIsolate();
}
isolate->set_long_jump_base(base);
Dart::ShutdownIsolate();
}
@@ -292,15 +293,15 @@ DEFINE_NATIVE_ENTRY(IsolateNatives_start, 2) {
// loaded, this check will throw an exception if they are not loaded.
if (init_successful && CheckArguments(library_url, class_name)) {
port_id = spawned_isolate->main_port();
uword data = reinterpret_cast<uword>(
new IsolateStartData(spawned_isolate,
strdup(library_url),
strdup(class_name),
port_id));
int result = Thread::Start(RunIsolate, data);
if (result != 0) {
FATAL1("Failed to start isolate thread %d", result);
}
spawned_isolate->set_spawn_data(
reinterpret_cast<uword>(
new IsolateStartData(strdup(library_url),
strdup(class_name),
port_id)));
Isolate::SetCurrent(NULL);
spawned_isolate->message_handler()->Run(
Dart::thread_pool(), RunIsolate, ShutdownIsolate,
reinterpret_cast<uword>(spawned_isolate));
} else {
// Error spawning the isolate, maybe due to initialization errors or
// errors while loading the application into spawned isolate, shut
@@ -425,10 +426,8 @@ class SpawnState {
}
void Cleanup() {
Isolate* saved = Isolate::Current();
Isolate::SetCurrent(isolate());
SwitchIsolateScope switch_scope(isolate());
Dart::ShutdownIsolate();
Isolate::SetCurrent(saved);
}
private:
@@ -479,16 +478,12 @@ static bool CreateIsolate(SpawnState* state, char** error) {
}
static void RunIsolate2(uword parameter) {
SpawnState* state = reinterpret_cast<SpawnState*>(parameter);
Isolate* isolate = state->isolate();
Isolate::SetCurrent(isolate);
// Intialize stack limit in case we are running isolate in a
// different thread than in which it was initialized.
isolate->SetStackLimitFromCurrentTOS(reinterpret_cast<uword>(&isolate));
static bool RunIsolate2(uword parameter) {
Isolate* isolate = reinterpret_cast<Isolate*>(parameter);
SpawnState* state = reinterpret_cast<SpawnState*>(isolate->spawn_data());
isolate->set_spawn_data(NULL);
{
StartIsolateScope start_scope(isolate);
Zone zone(isolate);
HandleScope handle_scope(isolate);
ASSERT(ClassFinalizer::FinalizePendingClasses());
@@ -503,16 +498,11 @@ static void RunIsolate2(uword parameter) {
const Array& kNoArgNames = Array::Handle();
result = DartEntry::InvokeStatic(func, args, kNoArgNames);
if (result.IsError()) {
ProcessError(result);
StoreError(isolate, result);
return false;
}
result = isolate->StandardRunLoop();
if (result.IsError()) {
ProcessError(result);
}
ASSERT(result.IsNull());
}
Dart::ShutdownIsolate();
return true;
}
@@ -551,15 +541,10 @@ DEFINE_NATIVE_ENTRY(isolate_spawnFunction, 1) {
}
// Start the new isolate.
int result = Thread::Start(RunIsolate2, reinterpret_cast<uword>(state));
if (result != 0) {
const String& msg = String::Handle(String::NewFormatted(
"Failed to start thread for isolate '%s'. Error code '%d'.",
state->isolate()->name(), result));
state->Cleanup();
delete state;
ThrowIsolateSpawnException(msg);
}
state->isolate()->set_spawn_data(reinterpret_cast<uword>(state));
state->isolate()->message_handler()->Run(
Dart::thread_pool(), RunIsolate2, ShutdownIsolate,
reinterpret_cast<uword>(state->isolate()));
arguments->SetReturn(port);
}
runtime/vm/code_generator.cc
+2 -36
View File
@@ -12,6 +12,7 @@
#include "vm/exceptions.h"
#include "vm/object_store.h"
#include "vm/message.h"
#include "vm/message_handler.h"
#include "vm/resolver.h"
#include "vm/runtime_entry.h"
#include "vm/stack_frame.h"
@@ -1125,20 +1126,6 @@ DEFINE_RUNTIME_ENTRY(ClosureArgumentMismatch, 0) {
}
static RawInstance* DeserializeMessage(void* data) {
// Create a snapshot object using the buffer.
const Snapshot* snapshot = Snapshot::SetupFromBuffer(data);
ASSERT(snapshot->IsMessageSnapshot());
// Read object back from the snapshot.
SnapshotReader reader(snapshot, Isolate::Current());
Instance& instance = Instance::Handle();
instance ^= reader.ReadObject();
return instance.raw();
}
DEFINE_RUNTIME_ENTRY(StackOverflow, 0) {
ASSERT(arguments.Count() ==
kStackOverflowRuntimeEntry.argument_count());
@@ -1157,28 +1144,7 @@ DEFINE_RUNTIME_ENTRY(StackOverflow, 0) {
uword interrupt_bits = isolate->GetAndClearInterrupts();
if (interrupt_bits & Isolate::kMessageInterrupt) {
while (true) {
// TODO(turnidge): This code is duplicated elsewhere. Consolidate.
Message* message =
isolate->message_handler()->queue()->DequeueNoWaitWithPriority(
Message::kOOBPriority);
if (message == NULL) {
// No more OOB messages to handle.
break;
}
const Instance& msg =
Instance::Handle(DeserializeMessage(message->data()));
// For now the only OOB messages are Mirrors messages.
const Object& result = Object::Handle(
DartLibraryCalls::HandleMirrorsMessage(
message->dest_port(), message->reply_port(), msg));
delete message;
if (result.IsError()) {
// TODO(turnidge): Propagating the error is probably wrong here.
Exceptions::PropagateError(result);
}
ASSERT(result.IsNull());
}
isolate->message_handler()->HandleOOBMessages();
}
if (interrupt_bits & Isolate::kApiInterrupt) {
Dart_IsolateInterruptCallback callback = isolate->InterruptCallback();
runtime/vm/dart.cc
+5
View File
@@ -15,6 +15,7 @@
#include "vm/port.h"
#include "vm/snapshot.h"
#include "vm/stub_code.h"
#include "vm/thread_pool.h"
#include "vm/virtual_memory.h"
#include "vm/zone.h"
@@ -23,8 +24,10 @@ namespace dart {
DECLARE_FLAG(bool, trace_isolates);
Isolate* Dart::vm_isolate_ = NULL;
ThreadPool* Dart::thread_pool_ = NULL;
DebugInfo* Dart::pprof_symbol_generator_ = NULL;
// TODO(turnidge): We should add a corresponding Dart::Cleanup.
bool Dart::InitOnce(Dart_IsolateCreateCallback create,
Dart_IsolateInterruptCallback interrupt) {
// TODO(iposva): Fix race condition here.
@@ -38,6 +41,8 @@ bool Dart::InitOnce(Dart_IsolateCreateCallback create,
FreeListElement::InitOnce();
Api::InitOnce();
// Create the VM isolate and finish the VM initialization.
ASSERT(thread_pool_ == NULL);
thread_pool_ = new ThreadPool();
{
ASSERT(vm_isolate_ == NULL);
ASSERT(Flags::Initialized());
runtime/vm/dart.h
+3
View File
@@ -14,6 +14,7 @@ namespace dart {
class DebugInfo;
class Isolate;
class RawError;
class ThreadPool;
class Dart : public AllStatic {
public:
@@ -25,6 +26,7 @@ class Dart : public AllStatic {
static void ShutdownIsolate();
static Isolate* vm_isolate() { return vm_isolate_; }
static ThreadPool* thread_pool() { return thread_pool_; }
static void set_pprof_symbol_generator(DebugInfo* value) {
pprof_symbol_generator_ = value;
@@ -33,6 +35,7 @@ class Dart : public AllStatic {
private:
static Isolate* vm_isolate_;
static ThreadPool* thread_pool_;
static DebugInfo* pprof_symbol_generator_;
};
runtime/vm/dart_api_impl.cc
+41 -53
View File
@@ -768,10 +768,42 @@ DART_EXPORT void Dart_SetMessageNotifyCallback(
}
struct RunLoopData {
Monitor* monitor;
bool done;
};
static void RunLoopDone(uword param) {
RunLoopData* data = reinterpret_cast<RunLoopData*>(param);
ASSERT(data->monitor != NULL);
MonitorLocker ml(data->monitor);
data->done = true;
ml.Notify();
}
DART_EXPORT Dart_Handle Dart_RunLoop() {
Isolate* isolate = Isolate::Current();
DARTSCOPE(isolate);
const Object& obj = Object::Handle(isolate, isolate->StandardRunLoop());
Monitor monitor;
MonitorLocker ml(&monitor);
{
SwitchIsolateScope switch_scope(NULL);
RunLoopData data;
data.monitor = &monitor;
data.done = false;
isolate->message_handler()->Run(
Dart::thread_pool(),
NULL, RunLoopDone, reinterpret_cast<uword>(&data));
while (!data.done) {
ml.Wait();
}
}
const Object& obj = Object::Handle(isolate->object_store()->sticky_error());
isolate->object_store()->clear_sticky_error();
if (obj.IsError()) {
return Api::NewHandle(isolate, obj.raw());
}
@@ -780,58 +812,13 @@ DART_EXPORT Dart_Handle Dart_RunLoop() {
}
static RawInstance* DeserializeMessage(Isolate* isolate, void* data) {
// Create a snapshot object using the buffer.
const Snapshot* snapshot = Snapshot::SetupFromBuffer(data);
ASSERT(snapshot->IsMessageSnapshot());
// Read object back from the snapshot.
SnapshotReader reader(snapshot, isolate);
Instance& instance = Instance::Handle(isolate);
instance ^= reader.ReadObject();
return instance.raw();
}
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);
// TODO(turnidge): This code is duplicated elsewhere. Consolidate.
message = isolate->message_handler()->queue()->DequeueNoWait();
if (message == NULL) {
break;
}
const Instance& msg =
Instance::Handle(isolate, DeserializeMessage(isolate, message->data()));
priority = message->priority();
if (priority == Message::kOOBPriority) {
// For now the only OOB messages are Mirrors messages.
const Object& result = Object::Handle(
isolate,
DartLibraryCalls::HandleMirrorsMessage(
message->dest_port(), message->reply_port(), msg));
delete message;
if (result.IsError()) {
// TODO(turnidge): Propagating the error is probably wrong here.
return Api::NewHandle(isolate, result.raw());
}
ASSERT(result.IsNull());
} else {
const Object& result = Object::Handle(
isolate,
DartLibraryCalls::HandleMessage(
message->dest_port(), message->reply_port(), msg));
delete message;
if (result.IsError()) {
return Api::NewHandle(isolate, result.raw());
}
ASSERT(result.IsNull());
}
} while (priority >= Message::kOOBPriority);
CHECK_ISOLATE(isolate);
if (!isolate->message_handler()->HandleNextMessage()) {
// TODO(turnidge): Clear sticky error here?
return Api::NewHandle(isolate, isolate->object_store()->sticky_error());
}
return Api::Success(isolate);
}
@@ -896,7 +883,8 @@ DART_EXPORT Dart_Port Dart_NewNativePort(const char* name,
name = "<UnnamedNativePort>";
}
if (handler == NULL) {
OS::PrintErr("%s expects argument 'handler' to be non-null.", CURRENT_FUNC);
OS::PrintErr("%s expects argument 'handler' to be non-null.\n",
CURRENT_FUNC);
return kIllegalPort;
}
// Start the native port without a current isolate.
@@ -905,7 +893,7 @@ DART_EXPORT Dart_Port Dart_NewNativePort(const char* name,
NativeMessageHandler* nmh = new NativeMessageHandler(name, handler);
Dart_Port port_id = PortMap::CreatePort(nmh);
nmh->StartWorker();
nmh->Run(Dart::thread_pool(), NULL, NULL, NULL);
return port_id;
}
runtime/vm/isolate.cc
+56 -63
View File
@@ -12,7 +12,7 @@
#include "vm/debugger.h"
#include "vm/debuginfo.h"
#include "vm/heap.h"
#include "vm/message.h"
#include "vm/message_handler.h"
#include "vm/object_store.h"
#include "vm/parser.h"
#include "vm/port.h"
@@ -39,6 +39,7 @@ class IsolateMessageHandler : public MessageHandler {
const char* name() const;
void MessageNotify(Message::Priority priority);
bool HandleMessage(Message* message);
#if defined(DEBUG)
// Check that it is safe to access this handler.
@@ -75,6 +76,57 @@ void IsolateMessageHandler::MessageNotify(Message::Priority priority) {
}
static RawInstance* DeserializeMessage(void* data) {
// Create a snapshot object using the buffer.
const Snapshot* snapshot = Snapshot::SetupFromBuffer(data);
ASSERT(snapshot->IsMessageSnapshot());
// Read object back from the snapshot.
SnapshotReader reader(snapshot, Isolate::Current());
Instance& instance = Instance::Handle();
instance ^= reader.ReadObject();
return instance.raw();
}
bool IsolateMessageHandler::HandleMessage(Message* message) {
StartIsolateScope start_scope(isolate_);
Zone zone(isolate_);
HandleScope handle_scope(isolate_);
const Instance& msg =
Instance::Handle(DeserializeMessage(message->data()));
if (message->IsOOB()) {
// For now the only OOB messages are Mirrors messages.
const Object& result = Object::Handle(
DartLibraryCalls::HandleMirrorsMessage(
message->dest_port(), message->reply_port(), msg));
delete message;
if (result.IsError()) {
// TODO(turnidge): Propagating the error is probably wrong here.
Error& error = Error::Handle();
error ^= result.raw();
isolate_->object_store()->set_sticky_error(error);
return false;
}
ASSERT(result.IsNull());
} else {
const Object& result = Object::Handle(
DartLibraryCalls::HandleMessage(
message->dest_port(), message->reply_port(), msg));
delete message;
if (result.IsError()) {
Error& error = Error::Handle();
error ^= result.raw();
isolate_->object_store()->set_sticky_error(error);
return false;
}
ASSERT(result.IsNull());
}
return true;
}
#if defined(DEBUG)
void IsolateMessageHandler::CheckAccess() {
ASSERT(isolate_ == Isolate::Current());
@@ -110,7 +162,9 @@ Isolate::Isolate()
ast_node_id_(AstNode::kNoId),
mutex_(new Mutex()),
stack_limit_(0),
saved_stack_limit_(0) {
saved_stack_limit_(0),
message_handler_(NULL),
spawn_data_(NULL) {
}
@@ -367,67 +421,6 @@ Dart_IsolateInterruptCallback Isolate::InterruptCallback() {
}
static RawInstance* DeserializeMessage(void* data) {
// Create a snapshot object using the buffer.
const Snapshot* snapshot = Snapshot::SetupFromBuffer(data);
ASSERT(snapshot->IsMessageSnapshot());
// Read object back from the snapshot.
SnapshotReader reader(snapshot, Isolate::Current());
Instance& instance = Instance::Handle();
instance ^= reader.ReadObject();
return instance.raw();
}
RawError* Isolate::StandardRunLoop() {
ASSERT(message_notify_callback() == NULL);
ASSERT(message_handler() != NULL);
while (message_handler()->HasLivePorts()) {
ASSERT(this == Isolate::Current());
Zone zone(this);
HandleScope handle_scope(this);
// TODO(turnidge): This code is duplicated elsewhere. Consolidate.
Message* message = message_handler()->queue()->Dequeue(0);
if (message != NULL) {
const Instance& msg =
Instance::Handle(DeserializeMessage(message->data()));
if (message->priority() >= Message::kOOBPriority) {
// For now the only OOB messages are Mirrors messages.
const Object& result = Object::Handle(
DartLibraryCalls::HandleMirrorsMessage(
message->dest_port(), message->reply_port(), msg));
delete message;
if (result.IsError()) {
// TODO(turnidge): Propagating the error is probably wrong here.
Error& error = Error::Handle();
error ^= result.raw();
return error.raw();
}
ASSERT(result.IsNull());
} else {
const Object& result = Object::Handle(
DartLibraryCalls::HandleMessage(
message->dest_port(), message->reply_port(), msg));
delete message;
if (result.IsError()) {
Error& error = Error::Handle();
error ^= result.raw();
return error.raw();
}
ASSERT(result.IsNull());
}
}
}
// Indicates success.
return Error::null();
}
void Isolate::VisitObjectPointers(ObjectPointerVisitor* visitor,
bool visit_prologue_weak_handles,
bool validate_frames) {
runtime/vm/isolate.h
+68 -5
View File
@@ -33,7 +33,6 @@ class StackResource;
class StubCode;
class Zone;
class Isolate : public BaseIsolate {
public:
~Isolate();
@@ -139,8 +138,7 @@ class Isolate : public BaseIsolate {
// value to trigger interrupts.
uword stack_limit() const { return stack_limit_; }
// The true stack limit for this isolate. This does not change
// after isolate initialization.
// The true stack limit for this isolate.
uword saved_stack_limit() const { return saved_stack_limit_; }
enum {
@@ -156,8 +154,8 @@ class Isolate : public BaseIsolate {
MessageHandler* message_handler() const { return message_handler_; }
void set_message_handler(MessageHandler* value) { message_handler_ = value; }
// Returns null on success, a RawError on failure.
RawError* StandardRunLoop();
uword spawn_data() const { return spawn_data_; }
void set_spawn_data(uword value) { spawn_data_ = value; }
intptr_t ast_node_id() const { return ast_node_id_; }
void set_ast_node_id(int value) { ast_node_id_ = value; }
@@ -211,6 +209,7 @@ class Isolate : public BaseIsolate {
uword stack_limit_;
uword saved_stack_limit_;
MessageHandler* message_handler_;
uword spawn_data_;
GcPrologueCallbacks gc_prologue_callbacks_;
GcEpilogueCallbacks gc_epilogue_callbacks_;
@@ -220,6 +219,70 @@ class Isolate : public BaseIsolate {
DISALLOW_COPY_AND_ASSIGN(Isolate);
};
// When we need to execute code in an isolate, we use the
// StartIsolateScope.
class StartIsolateScope {
public:
explicit StartIsolateScope(Isolate* new_isolate)
: new_isolate_(new_isolate), saved_isolate_(Isolate::Current()) {
ASSERT(new_isolate_ != NULL);
if (saved_isolate_ != new_isolate_) {
ASSERT(Isolate::Current() == NULL);
Isolate::SetCurrent(new_isolate_);
new_isolate_->SetStackLimitFromCurrentTOS(reinterpret_cast<uword>(this));
}
}
~StartIsolateScope() {
if (saved_isolate_ != new_isolate_) {
new_isolate_->SetStackLimit(~static_cast<uword>(0));
Isolate::SetCurrent(saved_isolate_);
}
}
private:
Isolate* new_isolate_;
Isolate* saved_isolate_;
DISALLOW_COPY_AND_ASSIGN(StartIsolateScope);
};
// When we need to temporarily become another isolate, we use the
// SwitchIsolateScope. It is not permitted to run dart code while in
// a SwitchIsolateScope.
class SwitchIsolateScope {
public:
explicit SwitchIsolateScope(Isolate* new_isolate)
: new_isolate_(new_isolate),
saved_isolate_(Isolate::Current()),
saved_stack_limit_(saved_isolate_
? saved_isolate_->saved_stack_limit() : 0) {
if (saved_isolate_ != new_isolate_) {
Isolate::SetCurrent(new_isolate_);
if (new_isolate_ != NULL) {
// Don't allow dart code to execute.
new_isolate_->SetStackLimit(~static_cast<uword>(0));
}
}
}
~SwitchIsolateScope() {
if (saved_isolate_ != new_isolate_) {
Isolate::SetCurrent(saved_isolate_);
if (saved_isolate_ != NULL) {
saved_isolate_->SetStackLimit(saved_stack_limit_);
}
}
}
private:
Isolate* new_isolate_;
Isolate* saved_isolate_;
uword saved_stack_limit_;
DISALLOW_COPY_AND_ASSIGN(SwitchIsolateScope);
};
} // namespace dart
#endif // VM_ISOLATE_H_
runtime/vm/message.cc
+44 -151
View File
@@ -8,193 +8,86 @@ namespace dart {
DECLARE_FLAG(bool, trace_isolates);
MessageHandler::MessageHandler()
: live_ports_(0),
queue_(new MessageQueue()) {
ASSERT(queue_ != NULL);
}
MessageHandler::~MessageHandler() {
delete queue_;
}
const char* MessageHandler::name() const {
return "<unnamed>";
}
#if defined(DEBUG)
void MessageHandler::CheckAccess() {
// By default there is no checking.
}
#endif
void MessageHandler::MessageNotify(Message::Priority priority) {
// By default, there is no custom message notification.
}
void MessageHandler::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::Priority priority = message->priority();
queue()->Enqueue(message);
message = NULL; // Do not access message. May have been deleted.
// Invoke any custom message notification.
MessageNotify(priority);
}
void MessageHandler::ClosePort(Dart_Port port) {
queue()->Flush(port);
}
void MessageHandler::CloseAllPorts() {
queue()->FlushAll();
}
MessageQueue::MessageQueue() {
for (int p = Message::kFirstPriority; p < Message::kNumPriorities; p++) {
head_[p] = NULL;
tail_[p] = NULL;
}
head_ = NULL;
tail_ = NULL;
}
MessageQueue::~MessageQueue() {
// Ensure that all pending messages have been released.
#if defined(DEBUG)
for (int p = Message::kFirstPriority; p < Message::kNumPriorities; p++) {
ASSERT(head_[p] == NULL);
}
ASSERT(head_ == NULL);
#endif
}
void MessageQueue::Enqueue(Message* msg) {
MonitorLocker ml(&monitor_);
Message::Priority p = msg->priority();
// Make sure messages are not reused.
ASSERT(msg->next_ == NULL);
if (head_[p] == NULL) {
if (head_ == NULL) {
// Only element in the queue.
head_[p] = msg;
tail_[p] = msg;
// We only need to notify if the queue was empty.
monitor_.Notify();
ASSERT(tail_ == NULL);
head_ = msg;
tail_ = msg;
} else {
ASSERT(tail_[p] != NULL);
ASSERT(tail_ != NULL);
// Append at the tail.
tail_[p]->next_ = msg;
tail_[p] = msg;
tail_->next_ = msg;
tail_ = msg;
}
}
Message* MessageQueue::DequeueNoWait() {
MonitorLocker ml(&monitor_);
return DequeueNoWaitHoldsLock(Message::kFirstPriority);
}
Message* MessageQueue::DequeueNoWaitWithPriority(
Message::Priority min_priority) {
MonitorLocker ml(&monitor_);
return DequeueNoWaitHoldsLock(min_priority);
}
Message* MessageQueue::DequeueNoWaitHoldsLock(Message::Priority min_priority) {
// Look for the highest priority available message.
for (int p = Message::kNumPriorities-1; p >= min_priority; p--) {
Message* result = head_[p];
if (result != NULL) {
head_[p] = result->next_;
// The following update to tail_ is not strictly needed.
if (head_[p] == NULL) {
tail_[p] = NULL;
}
#if defined(DEBUG)
result->next_ = result; // Make sure to trigger ASSERT in Enqueue.
#endif // DEBUG
return result;
Message* MessageQueue::Dequeue() {
Message* result = head_;
if (result != NULL) {
head_ = result->next_;
// The following update to tail_ is not strictly needed.
if (head_ == NULL) {
tail_ = NULL;
}
#if defined(DEBUG)
result->next_ = result; // Make sure to trigger ASSERT in Enqueue.
#endif // DEBUG
return result;
}
return NULL;
}
Message* MessageQueue::Dequeue(int64_t millis) {
ASSERT(millis >= 0);
MonitorLocker ml(&monitor_);
Message* result = DequeueNoWaitHoldsLock(Message::kFirstPriority);
if (result == NULL) {
// No message available at any priority.
monitor_.Wait(millis);
result = DequeueNoWaitHoldsLock(Message::kFirstPriority);
}
return result;
}
void MessageQueue::Flush(Dart_Port port) {
MonitorLocker ml(&monitor_);
for (int p = Message::kFirstPriority; p < Message::kNumPriorities; p++) {
Message* cur = head_[p];
Message* prev = NULL;
while (cur != NULL) {
Message* next = cur->next_;
// If the message matches, then remove it from the queue and delete it.
if (cur->dest_port() == port) {
if (prev != NULL) {
prev->next_ = next;
} else {
head_[p] = next;
}
delete cur;
Message* cur = head_;
Message* prev = NULL;
while (cur != NULL) {
Message* next = cur->next_;
// If the message matches, then remove it from the queue and delete it.
if (cur->dest_port() == port) {
if (prev != NULL) {
prev->next_ = next;
} else {
// Move prev forward.
prev = cur;
head_ = next;
}
// Advance to the next message in the queue.
cur = next;
delete cur;
} else {
// Move prev forward.
prev = cur;
}
tail_[p] = prev;
// Advance to the next message in the queue.
cur = next;
}
tail_ = prev;
}
void MessageQueue::FlushAll() {
MonitorLocker ml(&monitor_);
for (int p = Message::kFirstPriority; p < Message::kNumPriorities; p++) {
Message* cur = head_[p];
head_[p] = NULL;
tail_[p] = NULL;
while (cur != NULL) {
Message* next = cur->next_;
delete cur;
cur = next;
}
Message* cur = head_;
head_ = NULL;
tail_ = NULL;
while (cur != NULL) {
Message* next = cur->next_;
delete cur;
cur = next;
}
}
runtime/vm/message.h
+7 -73
View File
@@ -47,6 +47,8 @@ class Message {
uint8_t* data() const { return data_; }
Priority priority() const { return priority_; }
bool IsOOB() const { return priority_ == Message::kOOBPriority; }
private:
friend class MessageQueue;
@@ -67,20 +69,9 @@ class MessageQueue {
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.
Message* Dequeue(int64_t millis);
// Gets the next message from the message queue if available. Will
// not block.
Message* DequeueNoWait();
// Gets the next message of the specified priority or greater from
// the message queue if available. Will not block.
Message* DequeueNoWaitWithPriority(Message::Priority min_priority);
// Gets the next message from the message queue or NULL if no
// message is available. This function will not block.
Message* Dequeue();
void Flush(Dart_Port port);
void FlushAll();
@@ -88,69 +79,12 @@ class MessageQueue {
private:
friend class MessageQueueTestPeer;
Message* DequeueNoWaitHoldsLock(Message::Priority min_priority);
Monitor monitor_;
Message* head_[Message::kNumPriorities];
Message* tail_[Message::kNumPriorities];
Message* head_;
Message* tail_;
DISALLOW_COPY_AND_ASSIGN(MessageQueue);
};
// A MessageHandler is an entity capable of accepting messages.
class MessageHandler {
protected:
MessageHandler();
// Allows subclasses to provide custom message notification.
virtual void MessageNotify(Message::Priority priority);
public:
virtual ~MessageHandler();
// Allow subclasses to provide a handler name.
virtual const char* name() const;
#if defined(DEBUG)
// Check that it is safe to access this message handler.
//
// For example, if this MessageHandler is an isolate, then it is
// only safe to access it when the MessageHandler is the current
// isolate.
virtual void CheckAccess();
#endif
void PostMessage(Message* message);
void ClosePort(Dart_Port port);
void CloseAllPorts();
// A message handler tracks how many live ports it has.
bool HasLivePorts() const { return live_ports_ > 0; }
void increment_live_ports() {
#if defined(DEBUG)
CheckAccess();
#endif
live_ports_++;
}
void decrement_live_ports() {
#if defined(DEBUG)
CheckAccess();
#endif
live_ports_--;
}
// Returns true if the handler is owned by the PortMap.
//
// This is used to delete handlers when their last live port is closed.
virtual bool OwnedByPortMap() const { return false; }
MessageQueue* queue() const { return queue_; }
private:
intptr_t live_ports_;
MessageQueue* queue_;
};
} // namespace dart
#endif // VM_MESSAGE_H_
runtime/vm/message_handler.cc
+275
View File
@@ -0,0 +1,275 @@
// 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/message_handler.h"
#include "vm/dart.h"
namespace dart {
DECLARE_FLAG(bool, trace_isolates);
class MessageHandlerTask : public ThreadPool::Task {
public:
explicit MessageHandlerTask(MessageHandler* handler)
: handler_(handler) {
ASSERT(handler != NULL);
}
void Run() {
handler_->TaskCallback();
}
private:
MessageHandler* handler_;
DISALLOW_COPY_AND_ASSIGN(MessageHandlerTask);
};
MessageHandler::MessageHandler()
: queue_(new MessageQueue()),
oob_queue_(new MessageQueue()),
live_ports_(0),
pool_(NULL),
task_(NULL),
start_callback_(NULL),
end_callback_(NULL),
callback_data_(NULL) {
ASSERT(queue_ != NULL);
ASSERT(oob_queue_ != NULL);
}
MessageHandler::~MessageHandler() {
delete queue_;
delete oob_queue_;
}
const char* MessageHandler::name() const {
return "<unnamed>";
}
#if defined(DEBUG)
void MessageHandler::CheckAccess() {
// By default there is no checking.
}
#endif
void MessageHandler::MessageNotify(Message::Priority priority) {
// By default, there is no custom message notification.
}
void MessageHandler::Run(ThreadPool* pool,
StartCallback start_callback,
EndCallback end_callback,
CallbackData data) {
MonitorLocker ml(&monitor_);
if (FLAG_trace_isolates) {
OS::Print("[+] Starting message handler:\n"
"\thandler: %s\n",
name());
}
ASSERT(pool_ == NULL);
pool_ = pool;
start_callback_ = start_callback;
end_callback_ = end_callback;
callback_data_ = data;
task_ = new MessageHandlerTask(this);
pool_->Run(task_);
}
void MessageHandler::PostMessage(Message* message) {
MonitorLocker ml(&monitor_);
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::Priority saved_priority = message->priority();
if (message->IsOOB()) {
oob_queue_->Enqueue(message);
} else {
queue_->Enqueue(message);
}
message = NULL; // Do not access message. May have been deleted.
if (pool_ != NULL && task_ == NULL) {
task_ = new MessageHandlerTask(this);
pool_->Run(task_);
}
// Invoke any custom message notification.
MessageNotify(saved_priority);
}
Message* MessageHandler::DequeueMessage(Message::Priority min_priority) {
// TODO(turnidge): Add assert that monitor_ is held here.
Message* message = oob_queue_->Dequeue();
if (message == NULL && min_priority < Message::kOOBPriority) {
message = queue_->Dequeue();
}
return message;
}
bool MessageHandler::HandleMessages(bool allow_normal_messages,
bool allow_multiple_normal_messages) {
// TODO(turnidge): Add assert that monitor_ is held here.
bool result = true;
Message::Priority min_priority = (allow_normal_messages
? Message::kNormalPriority
: Message::kOOBPriority);
Message* message = DequeueMessage(min_priority);
while (message) {
if (FLAG_trace_isolates) {
OS::Print("[<] Handling message:\n"
"\thandler: %s\n"
"\tport: %lld\n",
name(), message->dest_port());
}
// Release the monitor_ temporarily while we handle the message.
// The monitor was acquired in MessageHandler::TaskCallback().
monitor_.Exit();
Message::Priority saved_priority = message->priority();
result = HandleMessage(message);
// ASSERT(Isolate::Current() == NULL);
monitor_.Enter();
if (!result) {
// If we hit an error, we're done processing messages.
break;
}
if (!allow_multiple_normal_messages &&
saved_priority == Message::kNormalPriority) {
// Some callers want to process only one normal message and then quit.
break;
}
message = DequeueMessage(min_priority);
}
return result;
}
bool MessageHandler::HandleNextMessage() {
// We can only call HandleNextMessage when this handler is not
// assigned to a thread pool.
MonitorLocker ml(&monitor_);
ASSERT(pool_ == NULL);
#if defined(DEBUG)
CheckAccess();
#endif
return HandleMessages(true, false);
}
bool MessageHandler::HandleOOBMessages() {
MonitorLocker ml(&monitor_);
#if defined(DEBUG)
CheckAccess();
#endif
return HandleMessages(false, false);
}
void MessageHandler::TaskCallback() {
ASSERT(Isolate::Current() == NULL);
bool ok = true;
bool run_end_callback = false;
{
MonitorLocker ml(&monitor_);
// Initialize the message handler by running its start function,
// if we have one. For an isolate, this will run the isolate's
// main() function.
if (start_callback_) {
monitor_.Exit();
ok = start_callback_(callback_data_);
ASSERT(Isolate::Current() == NULL);
start_callback_ = NULL;
monitor_.Enter();
}
// Handle any pending messages for this message handler.
if (ok) {
ok = HandleMessages(true, true);
}
task_ = NULL; // No task in queue.
if (!ok || !HasLivePorts()) {
if (FLAG_trace_isolates) {
OS::Print("[-] Stopping message handler (%s):\n"
"\thandler: %s\n",
(ok ? "no live ports" : "error"),
name());
}
pool_ = NULL;
run_end_callback = true;
}
}
if (run_end_callback && end_callback_ != NULL) {
end_callback_(callback_data_);
// The handler may have been deleted after this point.
}
}
void MessageHandler::ClosePort(Dart_Port port) {
MonitorLocker ml(&monitor_);
if (FLAG_trace_isolates) {
OS::Print("[-] Closing port:\n"
"\thandler: %s\n"
"\tport: %d\n",
name(), port);
}
queue_->Flush(port);
oob_queue_->Flush(port);
}
void MessageHandler::CloseAllPorts() {
MonitorLocker ml(&monitor_);
if (FLAG_trace_isolates) {
OS::Print("[-] Closing all ports:\n"
"\thandler: %s\n",
name());
}
queue_->FlushAll();
oob_queue_->FlushAll();
}
void MessageHandler::increment_live_ports() {
MonitorLocker ml(&monitor_);
#if defined(DEBUG)
CheckAccess();
#endif
live_ports_++;
}
void MessageHandler::decrement_live_ports() {
MonitorLocker ml(&monitor_);
#if defined(DEBUG)
CheckAccess();
#endif
live_ports_--;
}
} // namespace dart
runtime/vm/message_handler.h
+129
View File
@@ -0,0 +1,129 @@
// 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.
#ifndef VM_MESSAGE_HANDLER_H_
#define VM_MESSAGE_HANDLER_H_
#include "vm/message.h"
#include "vm/thread.h"
#include "vm/thread_pool.h"
namespace dart {
// A MessageHandler is an entity capable of accepting messages.
class MessageHandler {
protected:
MessageHandler();
public:
virtual ~MessageHandler();
// Allow subclasses to provide a handler name.
virtual const char* name() const;
typedef uword CallbackData;
typedef bool (*StartCallback)(CallbackData data);
typedef void (*EndCallback)(CallbackData data);
// Runs this message handler on the thread pool.
//
// Before processing messages, the optional StartFunction is run.
//
// A message handler will run until it terminates either normally or
// abnormally. Normal termination occurs when the message handler
// no longer has any live ports. Abnormal termination occurs when
// HandleMessage() indicates that an error has occurred during
// message processing.
void Run(ThreadPool* pool,
StartCallback start_callback,
EndCallback end_callback,
CallbackData data);
// Handles the next message for this message handler. Should only
// be used when not running the handler on the thread pool (via Run
// or RunBlocking).
//
// Returns true on success.
bool HandleNextMessage();
// Handles any OOB messages for this message handler. Can be used
// even if the message handler is running on the thread pool.
//
// Returns true on success.
bool HandleOOBMessages();
// A message handler tracks how many live ports it has.
bool HasLivePorts() const { return live_ports_ > 0; }
#if defined(DEBUG)
// Check that it is safe to access this message handler.
//
// For example, if this MessageHandler is an isolate, then it is
// only safe to access it when the MessageHandler is the current
// isolate.
virtual void CheckAccess();
#endif
protected:
// ------------ START PortMap API ------------
// These functions should only be called from the PortMap.
// Posts a message on this handler's message queue.
void PostMessage(Message* message);
// Notifies this handler that a port is being closed.
void ClosePort(Dart_Port port);
// Notifies this handler that all ports are being closed.
void CloseAllPorts();
// Returns true if the handler is owned by the PortMap.
//
// This is used to delete handlers when their last live port is closed.
virtual bool OwnedByPortMap() const { return false; }
void increment_live_ports();
void decrement_live_ports();
// ------------ END PortMap API ------------
// Custom message notification. Optionally provided by subclass.
virtual void MessageNotify(Message::Priority priority);
// Handles a single message. Provided by subclass.
//
// Returns true on success.
virtual bool HandleMessage(Message* message) = 0;
private:
friend class PortMap;
friend class MessageHandlerTestPeer;
friend class MessageHandlerTask;
// Called by MessageHandlerTask to process our task queue.
void TaskCallback();
// Dequeue the next message. Prefer messages from the oob_queue_ to
// messages from the queue_.
Message* DequeueMessage(Message::Priority min_priority);
// Handles any pending messages.
bool HandleMessages(bool allow_normal_messages,
bool allow_multiple_normal_messages);
Monitor monitor_; // Protects all fields in MessageHandler.
MessageQueue* queue_;
MessageQueue* oob_queue_;
intptr_t live_ports_;
ThreadPool* pool_;
ThreadPool::Task* task_;
StartCallback start_callback_;
EndCallback end_callback_;
CallbackData callback_data_;
DISALLOW_COPY_AND_ASSIGN(MessageHandler);
};
} // namespace dart
#endif // VM_MESSAGE_HANDLER_H_
runtime/vm/message_handler_test.cc
+268
View File
@@ -0,0 +1,268 @@
// 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) {}
void PostMessage(Message* message) { handler_->PostMessage(message); }
void ClosePort(Dart_Port port) { handler_->ClosePort(port); }
void CloseAllPorts() { handler_->CloseAllPorts(); }
void increment_live_ports() { handler_->increment_live_ports(); }
void decrement_live_ports() { handler_->decrement_live_ports(); }
MessageQueue* queue() const { return handler_->queue_; }
MessageQueue* oob_queue() const { return handler_->oob_queue_; }
private:
MessageHandler* handler_;
DISALLOW_COPY_AND_ASSIGN(MessageHandlerTestPeer);
};
class TestMessageHandler : public MessageHandler {
public:
TestMessageHandler()
: port_buffer_(strdup("")),
notify_count_(0),
message_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;
message_count_++;
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;
}
void End() {
intptr_t len =
OS::SNPrint(NULL, 0, "%s end", port_buffer_) + 1;
char* buffer = reinterpret_cast<char*>(malloc(len));
OS::SNPrint(buffer, len, "%s end", port_buffer_);
free(port_buffer_);
port_buffer_ = buffer;
}
const char* port_buffer() const { return port_buffer_; }
int notify_count() const { return notify_count_; }
int message_count() const { return message_count_; }
void set_result(bool result) { result_ = result; }
private:
char* port_buffer_;
int notify_count_;
int message_count_;
bool result_;
DISALLOW_COPY_AND_ASSIGN(TestMessageHandler);
};
bool TestStartFunction(uword data) {
return (reinterpret_cast<TestMessageHandler*>(data))->Start();
}
void TestEndFunction(uword data) {
return (reinterpret_cast<TestMessageHandler*>(data))->End();
}
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_peer.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_peer.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_peer.PostMessage(message1);
Message* message2 = new Message(2, 0, NULL, Message::kNormalPriority);
handler_peer.PostMessage(message2);
handler_peer.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_peer.PostMessage(message1);
Message* message2 = new Message(2, 0, NULL, Message::kNormalPriority);
handler_peer.PostMessage(message2);
handler_peer.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_peer.PostMessage(message1);
Message* oob_message1 = new Message(3, 0, NULL, Message::kOOBPriority);
handler_peer.PostMessage(oob_message1);
Message* message2 = new Message(2, 0, NULL, Message::kNormalPriority);
handler_peer.PostMessage(message2);
Message* oob_message2 = new Message(4, 0, NULL, Message::kOOBPriority);
handler_peer.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_peer.CloseAllPorts();
}
UNIT_TEST_CASE(MessageHandler_HandleOOBMessages) {
TestMessageHandler handler;
MessageHandlerTestPeer handler_peer(&handler);
Message* message1 = new Message(1, 0, NULL, Message::kNormalPriority);
handler_peer.PostMessage(message1);
Message* message2 = new Message(2, 0, NULL, Message::kNormalPriority);
handler_peer.PostMessage(message2);
Message* oob_message1 = new Message(3, 0, NULL, Message::kOOBPriority);
handler_peer.PostMessage(oob_message1);
Message* oob_message2 = new Message(4, 0, NULL, Message::kOOBPriority);
handler_peer.PostMessage(oob_message2);
// We handle both oob messages but no normal messages.
EXPECT(handler.HandleOOBMessages());
EXPECT_STREQ(" 3 4", handler.port_buffer());
handler_peer.CloseAllPorts();
}
struct ThreadStartInfo {
MessageHandler* handler;
int count;
};
static void SendMessages(uword param) {
ThreadStartInfo* info = reinterpret_cast<ThreadStartInfo*>(param);
MessageHandler* handler = info->handler;
MessageHandlerTestPeer handler_peer(handler);
for (int i = 0; i < info->count; i++) {
Message* message = new Message(i + 1, 0, NULL, Message::kNormalPriority);
handler_peer.PostMessage(message);
}
}
UNIT_TEST_CASE(MessageHandler_Run) {
ThreadPool pool;
TestMessageHandler handler;
MessageHandlerTestPeer handler_peer(&handler);
int sleep = 0;
const int kMaxSleep = 20 * 1000; // 20 seconds.
EXPECT(!handler.HasLivePorts());
handler_peer.increment_live_ports();
handler.Run(&pool,
TestStartFunction,
TestEndFunction,
reinterpret_cast<uword>(&handler));
Message* message = new Message(100, 0, NULL, Message::kNormalPriority);
handler_peer.PostMessage(message);
// Wait for the first message to be handled.
while (sleep < kMaxSleep && handler.message_count() < 1) {
OS::Sleep(10);
sleep += 10;
}
EXPECT_STREQ(" start 100", handler.port_buffer());
// Start a thread which sends more messages.
ThreadStartInfo info;
info.handler = &handler;
info.count = 10;
Thread::Start(SendMessages, reinterpret_cast<uword>(&info));
while (sleep < kMaxSleep && handler.message_count() < 11) {
OS::Sleep(10);
sleep += 10;
}
EXPECT_STREQ(" start 100 1 2 3 4 5 6 7 8 9 10", handler.port_buffer());
handler_peer.decrement_live_ports();
EXPECT(!handler.HasLivePorts());
}
} // namespace dart
runtime/vm/message_test.cc
+6 -148
View File
@@ -17,15 +17,14 @@ class MessageQueueTestPeer {
bool HasMessage() const {
// We don't really need to grab the monitor during the unit test,
// but it doesn't hurt.
queue_->monitor_.Enter();
bool result = (queue_->head_[Message::kNormalPriority] != NULL ||
queue_->head_[Message::kOOBPriority] != NULL);
queue_->monitor_.Exit();
bool result = (queue_->head_ != NULL);
return result;
}
private:
MessageQueue* queue_;
DISALLOW_COPY_AND_ASSIGN(MessageQueueTestPeer);
};
@@ -54,12 +53,12 @@ TEST_CASE(MessageQueue_BasicOperations) {
EXPECT(queue_peer.HasMessage());
// Remove two messages.
Message* msg = queue.Dequeue(0);
Message* msg = queue.Dequeue();
EXPECT(msg != NULL);
EXPECT_STREQ("msg1", reinterpret_cast<char*>(msg->data()));
EXPECT(queue_peer.HasMessage());
msg = queue.Dequeue(0);
msg = queue.Dequeue();
EXPECT(msg != NULL);
EXPECT_STREQ("msg2", reinterpret_cast<char*>(msg->data()));
EXPECT(!queue_peer.HasMessage());
@@ -69,147 +68,6 @@ TEST_CASE(MessageQueue_BasicOperations) {
}
TEST_CASE(MessageQueue_Priorities) {
MessageQueue queue;
MessageQueueTestPeer queue_peer(&queue);
EXPECT(!queue_peer.HasMessage());
Dart_Port port = 1;
// Add two messages.
Message* msg1 =
new Message(port, 0, AllocMsg("msg1"), Message::kNormalPriority);
queue.Enqueue(msg1);
EXPECT(queue_peer.HasMessage());
Message* msg2 =
new Message(port, 0, AllocMsg("msg2"), Message::kOOBPriority);
queue.Enqueue(msg2);
EXPECT(queue_peer.HasMessage());
// The higher priority message is delivered first.
Message* msg = queue.Dequeue(0);
EXPECT(msg != NULL);
EXPECT_STREQ("msg2", reinterpret_cast<char*>(msg->data()));
EXPECT(queue_peer.HasMessage());
msg = queue.Dequeue(0);
EXPECT(msg != NULL);
EXPECT_STREQ("msg1", reinterpret_cast<char*>(msg->data()));
EXPECT(!queue_peer.HasMessage());
delete msg1;
delete msg2;
}
// A thread which receives an expected sequence of messages.
static Monitor* sync = NULL;
static MessageQueue* shared_queue = NULL;
void MessageReceiver_start(uword unused) {
// We only need an isolate here because the MonitorLocker in the
// MessageQueue expects it, we don't need to initialize the isolate
// as it does not run any dart code.
Dart::CreateIsolate(NULL);
// Create a message queue and share it.
MessageQueue* queue = new MessageQueue();
MessageQueueTestPeer peer(queue);
shared_queue = queue;
// Tell the other thread that the shared queue is ready.
{
MonitorLocker ml(sync);
ml.Notify();
}
// Wait for the other thread to fill the queue a bit.
while (!peer.HasMessage()) {
MonitorLocker ml(sync);
ml.Wait(5);
}
int i = 0;
while (i < 3) {
Message* msg = queue->Dequeue(0);
// Dequeue(0) can return NULL due to spurious wakeup.
if (msg != NULL) {
EXPECT_EQ(i + 10, msg->dest_port());
EXPECT_EQ(i + 100, msg->reply_port());
EXPECT_EQ(i + 1000, *(reinterpret_cast<int*>(msg->data())));
delete msg;
i++;
}
}
i = 0;
while (i < 3) {
Message* msg = queue->Dequeue(0);
// Dequeue(0) can return NULL due to spurious wakeup.
if (msg != NULL) {
EXPECT_EQ(i + 20, msg->dest_port());
EXPECT_EQ(i + 200, msg->reply_port());
EXPECT_EQ(i + 2000, *(reinterpret_cast<int*>(msg->data())));
delete msg;
i++;
}
}
shared_queue = NULL;
delete queue;
Dart::ShutdownIsolate();
}
TEST_CASE(MessageQueue_WaitNotify) {
sync = new Monitor();
int result = Thread::Start(MessageReceiver_start, 0);
EXPECT_EQ(0, result);
// Wait for the shared queue to be created.
while (shared_queue == NULL) {
MonitorLocker ml(sync);
ml.Wait(5);
}
ASSERT(shared_queue != NULL);
// Pile up three messages before the other thread runs.
for (int i = 0; i < 3; i++) {
int* data = reinterpret_cast<int*>(malloc(sizeof(*data)));
*data = i + 1000;
Message* msg =
new Message(i + 10, i + 100, reinterpret_cast<uint8_t*>(data),
Message::kNormalPriority);
shared_queue->Enqueue(msg);
}
// Wake the other thread and have it start consuming messages.
{
MonitorLocker ml(sync);
ml.Notify();
}
// Add a few more messages after sleeping to allow the other thread
// to potentially exercise the blocking code path in Dequeue.
OS::Sleep(5);
for (int i = 0; i < 3; i++) {
int* data = reinterpret_cast<int*>(malloc(sizeof(*data)));
*data = i + 2000;
Message* msg =
new Message(i + 20, i + 200, reinterpret_cast<uint8_t*>(data),
Message::kNormalPriority);
shared_queue->Enqueue(msg);
}
sync = NULL;
delete sync;
// Give the spawned thread enough time to properly exit.
OS::Sleep(20);
}
TEST_CASE(MessageQueue_FlushAll) {
MessageQueue queue;
MessageQueueTestPeer queue_peer(&queue);
@@ -251,7 +109,7 @@ TEST_CASE(MessageQueue_Flush) {
// One message is left in the queue.
EXPECT(queue_peer.HasMessage());
Message* msg = queue.Dequeue(0);
Message* msg = queue.Dequeue();
EXPECT(msg != NULL);
EXPECT_STREQ("msg2", reinterpret_cast<char*>(msg->data()));
runtime/vm/native_message_handler.cc
+17 -40
View File
@@ -41,47 +41,24 @@ static uint8_t* zone_allocator(
}
static void RunWorker(uword parameter) {
NativeMessageHandler* handler =
reinterpret_cast<NativeMessageHandler*>(parameter);
#if defined(DEBUG)
handler->CheckAccess();
#endif
while (handler->HasLivePorts()) {
Message* message = handler->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();
}
// Enter a native scope for handling the message. This will create a
// zone for allocating the objects for decoding the message.
ApiNativeScope scope;
int32_t length = reinterpret_cast<int32_t*>(
message->data())[Snapshot::kLengthIndex];
ApiMessageReader reader(message->data() + Snapshot::kHeaderSize,
length,
zone_allocator);
Dart_CObject* object = reader.ReadMessage();
(*handler->func())(message->dest_port(),
message->reply_port(),
object);
delete message;
}
bool NativeMessageHandler::HandleMessage(Message* message) {
if (message->IsOOB()) {
// We currently do not use OOB messages for native ports.
UNREACHABLE();
}
// Enter a native scope for handling the message. This will create a
// zone for allocating the objects for decoding the message.
ApiNativeScope scope;
int32_t length = reinterpret_cast<int32_t*>(
message->data())[Snapshot::kLengthIndex];
ApiMessageReader reader(message->data() + Snapshot::kHeaderSize,
length,
zone_allocator);
Dart_CObject* object = reader.ReadMessage();
(*func())(message->dest_port(), message->reply_port(), object);
delete message;
return true;
}
void NativeMessageHandler::StartWorker() {
int result = Thread::Start(RunWorker, reinterpret_cast<uword>(this));
if (result != 0) {
FATAL1("Failed to start native message handler worker thread %d", result);
}
}
} // namespace dart
runtime/vm/native_message_handler.h
+3 -10
View File
@@ -6,7 +6,7 @@
#define VM_NATIVE_MESSAGE_HANDLER_H_
#include "include/dart_api.h"
#include "vm/message.h"
#include "vm/message_handler.h"
namespace dart {
@@ -20,6 +20,8 @@ class NativeMessageHandler : public MessageHandler {
const char* name() const { return name_; }
Dart_NativeMessageHandler func() const { return func_; }
bool HandleMessage(Message* message);
#if defined(DEBUG)
// Check that it is safe to access this handler.
void CheckAccess();
@@ -28,15 +30,6 @@ class NativeMessageHandler : public MessageHandler {
// Delete this handlers when its last live port is closed.
virtual bool OwnedByPortMap() const { return true; }
// Start a worker thread which will service messages for this handler.
//
// TODO(turnidge): Instead of starting a worker for each
// NativeMessageHandler, we should instead use a shared thread pool
// which services a queue of ready MessageHandlers. If we implement
// this correctly, the same pool will work for
// IsolateMessageHandlers as well.
void StartWorker();
private:
char* name_;
Dart_NativeMessageHandler func_;
runtime/vm/port.cc
+1 -1
View File
@@ -7,7 +7,7 @@
#include "platform/utils.h"
#include "vm/dart_api_impl.h"
#include "vm/isolate.h"
#include "vm/message.h"
#include "vm/message_handler.h"
#include "vm/thread.h"
namespace dart {
runtime/vm/port_test.cc
+11 -103
View File
@@ -3,7 +3,7 @@
// BSD-style license that can be found in the LICENSE file.
#include "platform/assert.h"
#include "vm/message.h"
#include "vm/message_handler.h"
#include "vm/os.h"
#include "vm/port.h"
#include "vm/unit_test.h"
@@ -29,20 +29,22 @@ class PortMapTestPeer {
};
class TestMessageHandler : public MessageHandler {
class PortTestMessageHandler : public MessageHandler {
public:
TestMessageHandler() : notify_count(0) {}
PortTestMessageHandler() : notify_count(0) {}
void MessageNotify(Message::Priority priority) {
notify_count++;
}
bool HandleMessage(Message* message) { return true; }
int notify_count;
};
TEST_CASE(PortMap_CreateAndCloseOnePort) {
TestMessageHandler handler;
PortTestMessageHandler handler;
intptr_t port = PortMap::CreatePort(&handler);
EXPECT_NE(0, port);
EXPECT(PortMapTestPeer::IsActivePort(port));
@@ -53,7 +55,7 @@ TEST_CASE(PortMap_CreateAndCloseOnePort) {
TEST_CASE(PortMap_CreateAndCloseTwoPorts) {
TestMessageHandler handler;
PortTestMessageHandler handler;
Dart_Port port1 = PortMap::CreatePort(&handler);
Dart_Port port2 = PortMap::CreatePort(&handler);
EXPECT(PortMapTestPeer::IsActivePort(port1));
@@ -73,7 +75,7 @@ TEST_CASE(PortMap_CreateAndCloseTwoPorts) {
TEST_CASE(PortMap_ClosePorts) {
TestMessageHandler handler;
PortTestMessageHandler handler;
Dart_Port port1 = PortMap::CreatePort(&handler);
Dart_Port port2 = PortMap::CreatePort(&handler);
EXPECT(PortMapTestPeer::IsActivePort(port1));
@@ -87,7 +89,7 @@ TEST_CASE(PortMap_ClosePorts) {
TEST_CASE(PortMap_CreateManyPorts) {
TestMessageHandler handler;
PortTestMessageHandler handler;
for (int i = 0; i < 32; i++) {
Dart_Port port = PortMap::CreatePort(&handler);
EXPECT(PortMapTestPeer::IsActivePort(port));
@@ -98,7 +100,7 @@ TEST_CASE(PortMap_CreateManyPorts) {
TEST_CASE(PortMap_SetLive) {
TestMessageHandler handler;
PortTestMessageHandler handler;
intptr_t port = PortMap::CreatePort(&handler);
EXPECT_NE(0, port);
EXPECT(PortMapTestPeer::IsActivePort(port));
@@ -115,7 +117,7 @@ TEST_CASE(PortMap_SetLive) {
TEST_CASE(PortMap_PostMessage) {
TestMessageHandler handler;
PortTestMessageHandler handler;
Dart_Port port = PortMap::CreatePort(&handler);
EXPECT_EQ(0, handler.notify_count);
@@ -135,98 +137,4 @@ TEST_CASE(PortMap_PostMessageInvalidPort) {
Message::kNormalPriority)));
}
// End-of-test marker.
static const intptr_t kEOT = 0xFFFF;
uint8_t* AllocIntData(intptr_t payload) {
intptr_t* result = reinterpret_cast<intptr_t*>(malloc(sizeof(payload)));
*result = payload;
return reinterpret_cast<uint8_t*>(result);
}
intptr_t GetIntData(uint8_t* data) {
return *reinterpret_cast<intptr_t*>(data);
}
static Message* NextMessage() {
Isolate* isolate = Isolate::Current();
Message* result = isolate->message_handler()->queue()->Dequeue(0);
return result;
}
void ThreadedPort_start(uword parameter) {
// TODO(turnidge): We only use the isolate to get access to its
// message handler. I should rewrite this test to use a
// TestMessageHandler instead.
Isolate* isolate = Dart::CreateIsolate(NULL);
intptr_t remote = parameter;
intptr_t local = PortMap::CreatePort(isolate->message_handler());
PortMap::PostMessage(new Message(
remote, 0, AllocIntData(local), Message::kNormalPriority));
intptr_t count = 0;
while (true) {
Message* msg = NextMessage();
EXPECT_EQ(local, msg->dest_port());
EXPECT(msg != NULL);
if (GetIntData(msg->data()) == kEOT) {
break;
}
EXPECT(GetIntData(msg->data()) == count);
delete msg;
PortMap::PostMessage(new Message(
remote, 0, AllocIntData(count * 2), Message::kNormalPriority));
count++;
}
PortMap::PostMessage(new Message(
remote, 0, AllocIntData(kEOT), Message::kNormalPriority));
Dart::ShutdownIsolate();
}
TEST_CASE(ThreadedPort) {
intptr_t local = PortMap::CreatePort(Isolate::Current()->message_handler());
int result = Thread::Start(ThreadedPort_start, local);
EXPECT_EQ(0, result);
Message* msg = NextMessage();
EXPECT_EQ(local, msg->dest_port());
EXPECT(msg != NULL);
intptr_t remote = GetIntData(msg->data()); // Get the remote port.
delete msg;
for (intptr_t i = 0; i < 10; i++) {
PortMap::PostMessage(
new Message(remote, 0, AllocIntData(i), Message::kNormalPriority));
Message* msg = NextMessage();
EXPECT_EQ(local, msg->dest_port());
EXPECT(msg != NULL);
EXPECT_EQ(i * 2, GetIntData(msg->data()));
delete msg;
}
PortMap::PostMessage(
new Message(remote, 0, AllocIntData(kEOT), Message::kNormalPriority));
msg = NextMessage();
EXPECT_EQ(local, msg->dest_port());
EXPECT(msg != NULL);
EXPECT_EQ(kEOT, GetIntData(msg->data()));
delete msg;
// Give the spawned thread enough time to properly exit.
Monitor* waiter = new Monitor();
{
MonitorLocker ml(waiter);
ml.Wait(20);
}
delete waiter;
}
} // namespace dart
runtime/vm/thread_pool.cc
+4 -1
View File
@@ -228,7 +228,10 @@ void ThreadPool::Worker::StartThread() {
ASSERT(task_ != NULL);
}
#endif
Thread::Start(&Worker::Main, reinterpret_cast<uword>(this));
int result = Thread::Start(&Worker::Main, reinterpret_cast<uword>(this));
if (result != 0) {
FATAL1("Could not start worker thread: result = %d.", result);
}
}
runtime/vm/vm_sources.gypi
+3
View File
@@ -171,6 +171,9 @@
'message.cc',
'message.h',
'message_test.cc',
'message_handler.cc',
'message_handler.h',
'message_handler_test.cc',
'native_arguments.cc',
'native_arguments.h',
'native_entry.cc',