diff --git a/runtime/lib/isolate.cc b/runtime/lib/isolate.cc index 3239954c3f3..f2498aa08f5 100644 --- a/runtime/lib/isolate.cc +++ b/runtime/lib/isolate.cc @@ -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(parameter); - Isolate* isolate = data->isolate_; +static bool RunIsolate(uword parameter) { + Isolate* isolate = reinterpret_cast(parameter); + IsolateStartData* data = + reinterpret_cast(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(&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 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(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( - 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( + 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(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(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(&isolate)); - +static bool RunIsolate2(uword parameter) { + Isolate* isolate = reinterpret_cast(parameter); + SpawnState* state = reinterpret_cast(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(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(state)); + state->isolate()->message_handler()->Run( + Dart::thread_pool(), RunIsolate2, ShutdownIsolate, + reinterpret_cast(state->isolate())); arguments->SetReturn(port); } diff --git a/runtime/vm/code_generator.cc b/runtime/vm/code_generator.cc index aa18161762d..425166aea96 100644 --- a/runtime/vm/code_generator.cc +++ b/runtime/vm/code_generator.cc @@ -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(); diff --git a/runtime/vm/dart.cc b/runtime/vm/dart.cc index fe779217035..fadeb3891be 100644 --- a/runtime/vm/dart.cc +++ b/runtime/vm/dart.cc @@ -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()); diff --git a/runtime/vm/dart.h b/runtime/vm/dart.h index a7f943e2a9b..fe640ea28d0 100644 --- a/runtime/vm/dart.h +++ b/runtime/vm/dart.h @@ -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_; }; diff --git a/runtime/vm/dart_api_impl.cc b/runtime/vm/dart_api_impl.cc index 4e7632155a9..d58252d0d38 100644 --- a/runtime/vm/dart_api_impl.cc +++ b/runtime/vm/dart_api_impl.cc @@ -768,10 +768,42 @@ DART_EXPORT void Dart_SetMessageNotifyCallback( } +struct RunLoopData { + Monitor* monitor; + bool done; +}; + + +static void RunLoopDone(uword param) { + RunLoopData* data = reinterpret_cast(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(&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 = ""; } 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; } diff --git a/runtime/vm/isolate.cc b/runtime/vm/isolate.cc index 32f54546f81..b38b63e6635 100644 --- a/runtime/vm/isolate.cc +++ b/runtime/vm/isolate.cc @@ -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) { diff --git a/runtime/vm/isolate.h b/runtime/vm/isolate.h index 448b2b37ebc..063ffb460b4 100644 --- a/runtime/vm/isolate.h +++ b/runtime/vm/isolate.h @@ -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(this)); + } + } + + ~StartIsolateScope() { + if (saved_isolate_ != new_isolate_) { + new_isolate_->SetStackLimit(~static_cast(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(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_ diff --git a/runtime/vm/message.cc b/runtime/vm/message.cc index 64e10d22d62..7051bd9348c 100644 --- a/runtime/vm/message.cc +++ b/runtime/vm/message.cc @@ -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 ""; -} - - -#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 = ""; - 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; } } diff --git a/runtime/vm/message.h b/runtime/vm/message.h index c98bd003e16..252a1453be2 100644 --- a/runtime/vm/message.h +++ b/runtime/vm/message.h @@ -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_ diff --git a/runtime/vm/message_handler.cc b/runtime/vm/message_handler.cc new file mode 100644 index 00000000000..56f79a05b2a --- /dev/null +++ b/runtime/vm/message_handler.cc @@ -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 ""; +} + + +#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 = ""; + 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 diff --git a/runtime/vm/message_handler.h b/runtime/vm/message_handler.h new file mode 100644 index 00000000000..acbca5b27ec --- /dev/null +++ b/runtime/vm/message_handler.h @@ -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_ diff --git a/runtime/vm/message_handler_test.cc b/runtime/vm/message_handler_test.cc new file mode 100644 index 00000000000..89a2563a1ac --- /dev/null +++ b/runtime/vm/message_handler_test.cc @@ -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(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(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(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(data))->Start(); +} + + +void TestEndFunction(uword data) { + return (reinterpret_cast(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(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(&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(&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 diff --git a/runtime/vm/message_test.cc b/runtime/vm/message_test.cc index 6f9cbe157da..5e729b24f73 100644 --- a/runtime/vm/message_test.cc +++ b/runtime/vm/message_test.cc @@ -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(msg->data())); EXPECT(queue_peer.HasMessage()); - msg = queue.Dequeue(0); + msg = queue.Dequeue(); EXPECT(msg != NULL); EXPECT_STREQ("msg2", reinterpret_cast(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(msg->data())); - EXPECT(queue_peer.HasMessage()); - - msg = queue.Dequeue(0); - EXPECT(msg != NULL); - EXPECT_STREQ("msg1", reinterpret_cast(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(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(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(malloc(sizeof(*data))); - *data = i + 1000; - Message* msg = - new Message(i + 10, i + 100, reinterpret_cast(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(malloc(sizeof(*data))); - *data = i + 2000; - Message* msg = - new Message(i + 20, i + 200, reinterpret_cast(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(msg->data())); diff --git a/runtime/vm/native_message_handler.cc b/runtime/vm/native_message_handler.cc index 8f47e7ff748..9b34df7f6fc 100644 --- a/runtime/vm/native_message_handler.cc +++ b/runtime/vm/native_message_handler.cc @@ -41,47 +41,24 @@ static uint8_t* zone_allocator( } -static void RunWorker(uword parameter) { - NativeMessageHandler* handler = - reinterpret_cast(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( - 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( + 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(this)); - if (result != 0) { - FATAL1("Failed to start native message handler worker thread %d", result); - } -} - - } // namespace dart diff --git a/runtime/vm/native_message_handler.h b/runtime/vm/native_message_handler.h index 8334fb53ab4..d3f29a8fefe 100644 --- a/runtime/vm/native_message_handler.h +++ b/runtime/vm/native_message_handler.h @@ -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_; diff --git a/runtime/vm/port.cc b/runtime/vm/port.cc index 15590e62fdf..5bcc6bcc78b 100644 --- a/runtime/vm/port.cc +++ b/runtime/vm/port.cc @@ -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 { diff --git a/runtime/vm/port_test.cc b/runtime/vm/port_test.cc index b855cdb2ea6..ab4510260de 100644 --- a/runtime/vm/port_test.cc +++ b/runtime/vm/port_test.cc @@ -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(malloc(sizeof(payload))); - *result = payload; - return reinterpret_cast(result); -} - - -intptr_t GetIntData(uint8_t* data) { - return *reinterpret_cast(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 diff --git a/runtime/vm/thread_pool.cc b/runtime/vm/thread_pool.cc index cdb094582a3..ab21ebd2827 100644 --- a/runtime/vm/thread_pool.cc +++ b/runtime/vm/thread_pool.cc @@ -228,7 +228,10 @@ void ThreadPool::Worker::StartThread() { ASSERT(task_ != NULL); } #endif - Thread::Start(&Worker::Main, reinterpret_cast(this)); + int result = Thread::Start(&Worker::Main, reinterpret_cast(this)); + if (result != 0) { + FATAL1("Could not start worker thread: result = %d.", result); + } } diff --git a/runtime/vm/vm_sources.gypi b/runtime/vm/vm_sources.gypi index e1d6e35f964..385e4af4d32 100644 --- a/runtime/vm/vm_sources.gypi +++ b/runtime/vm/vm_sources.gypi @@ -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',