From c48496b8735c1bde19dfbf87cefe1cdfe8ee788d Mon Sep 17 00:00:00 2001 From: "turnidge@google.com" Date: Wed, 20 Aug 2014 19:57:09 +0000 Subject: [PATCH] Refactor isolate startup code in preparation for making isolate spawning more truly non-blocking. Instead of passing a startup message to the new isolate, pass all necessary information in to _startIsolate directly. The new isolate sends its control port and capabilities back to the parent. R=iposva@google.com Review URL: https://codereview.chromium.org//456983002 git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@39423 260f80e4-7a28-3924-810f-c04153c831b5 --- runtime/bin/main.cc | 27 +++---- runtime/lib/isolate.cc | 42 ++++------- runtime/lib/isolate_patch.dart | 130 ++++++++++++++++++--------------- runtime/vm/bootstrap_natives.h | 7 +- runtime/vm/isolate.cc | 93 ++++++++++++++++++++--- runtime/vm/isolate.h | 18 ++++- runtime/vm/message_handler.cc | 23 +----- runtime/vm/message_handler.h | 14 +--- runtime/vm/object.cc | 10 ++- runtime/vm/object.h | 4 +- runtime/vm/port.cc | 34 +++++++-- runtime/vm/port.h | 12 ++- runtime/vm/port_test.cc | 22 +++++- 13 files changed, 265 insertions(+), 171 deletions(-) diff --git a/runtime/bin/main.cc b/runtime/bin/main.cc index 46f0652268d..4c3c7dd3aa5 100644 --- a/runtime/bin/main.cc +++ b/runtime/bin/main.cc @@ -1125,29 +1125,22 @@ void main(int argc, char** argv) { // Call _startIsolate in the isolate library to enable dispatching the // initial startup message. - Dart_Handle isolate_args[2]; - isolate_args[0] = main_closure; - isolate_args[1] = Dart_True(); + const intptr_t kNumIsolateArgs = 7; + Dart_Handle isolate_args[kNumIsolateArgs]; + isolate_args[0] = Dart_Null(); // no parent port + isolate_args[1] = main_closure; // entryPoint + isolate_args[2] = CreateRuntimeOptions(&dart_options); // args + isolate_args[3] = Dart_Null(); // no message + isolate_args[4] = Dart_True(); // isSpawnUri + isolate_args[5] = Dart_Null(); // no control port + isolate_args[6] = Dart_Null(); // no capabilities Dart_Handle isolate_lib = Dart_LookupLibrary( Dart_NewStringFromCString("dart:isolate")); result = Dart_Invoke(isolate_lib, Dart_NewStringFromCString("_startIsolate"), - 2, isolate_args); - - // Setup the arguments in the initial startup message and leave the - // replyTo and message fields empty. - Dart_Handle initial_startup_msg = Dart_NewList(3); - result = Dart_ListSetAt(initial_startup_msg, 1, - CreateRuntimeOptions(&dart_options)); + kNumIsolateArgs, isolate_args); DartExitOnError(result); - Dart_Port main_port = Dart_GetMainPortId(); - bool posted = Dart_Post(main_port, initial_startup_msg); - if (!posted) { - ErrorExit(kErrorExitCode, - "Failed posting startup message to main " - "isolate control port."); - } // Keep handling messages until the last active receive port is closed. result = Dart_RunLoop(); diff --git a/runtime/lib/isolate.cc b/runtime/lib/isolate.cc index 73b611489d9..f85726d1e02 100644 --- a/runtime/lib/isolate.cc +++ b/runtime/lib/isolate.cc @@ -38,7 +38,7 @@ DEFINE_NATIVE_ENTRY(RawReceivePortImpl_factory, 1) { ASSERT(TypeArguments::CheckedHandle(arguments->NativeArgAt(0)).IsNull()); Dart_Port port_id = PortMap::CreatePort(arguments->isolate()->message_handler()); - return ReceivePort::New(port_id); + return ReceivePort::New(port_id, false /* not control port */); } @@ -177,20 +177,10 @@ static RawObject* Spawn(Isolate* parent_isolate, ThrowIsolateSpawnException(msg); } - // The result of spawning an Isolate is an array with 3 elements: - // [main_port, pause_capability, terminate_capability] - const Array& result = Array::Handle(Array::New(3)); - // Create a SendPort for the new isolate. Isolate* spawned_isolate = state->isolate(); const SendPort& port = SendPort::Handle( SendPort::New(spawned_isolate->main_port())); - result.SetAt(0, port); - Capability& capability = Capability::Handle(); - capability = Capability::New(spawned_isolate->pause_capability()); - result.SetAt(1, capability); // pauseCapability - capability = Capability::New(spawned_isolate->terminate_capability()); - result.SetAt(2, capability); // terminateCapability // Start the new isolate if it is already marked as runnable. MutexLocker ml(spawned_isolate->mutex()); @@ -199,12 +189,14 @@ static RawObject* Spawn(Isolate* parent_isolate, spawned_isolate->Run(); } - return result.raw(); + return port.raw(); } -DEFINE_NATIVE_ENTRY(Isolate_spawnFunction, 1) { - GET_NON_NULL_NATIVE_ARGUMENT(Instance, closure, arguments->NativeArgAt(0)); +DEFINE_NATIVE_ENTRY(Isolate_spawnFunction, 3) { + GET_NON_NULL_NATIVE_ARGUMENT(SendPort, port, arguments->NativeArgAt(0)); + GET_NON_NULL_NATIVE_ARGUMENT(Instance, closure, arguments->NativeArgAt(1)); + GET_NON_NULL_NATIVE_ARGUMENT(Instance, message, arguments->NativeArgAt(2)); if (closure.IsClosure()) { Function& func = Function::Handle(); func = Closure::function(closure); @@ -214,7 +206,7 @@ DEFINE_NATIVE_ENTRY(Isolate_spawnFunction, 1) { ctx = Closure::context(closure); ASSERT(ctx.num_variables() == 0); #endif - return Spawn(isolate, new IsolateSpawnState(func)); + return Spawn(isolate, new IsolateSpawnState(port.Id(), func, message)); } } const String& msg = String::Handle(String::New( @@ -224,8 +216,11 @@ DEFINE_NATIVE_ENTRY(Isolate_spawnFunction, 1) { } -DEFINE_NATIVE_ENTRY(Isolate_spawnUri, 1) { - GET_NON_NULL_NATIVE_ARGUMENT(String, uri, arguments->NativeArgAt(0)); +DEFINE_NATIVE_ENTRY(Isolate_spawnUri, 4) { + GET_NON_NULL_NATIVE_ARGUMENT(SendPort, port, arguments->NativeArgAt(0)); + GET_NON_NULL_NATIVE_ARGUMENT(String, uri, arguments->NativeArgAt(1)); + GET_NON_NULL_NATIVE_ARGUMENT(Instance, args, arguments->NativeArgAt(2)); + GET_NON_NULL_NATIVE_ARGUMENT(Instance, message, arguments->NativeArgAt(3)); // Canonicalize the uri with respect to the current isolate. char* error = NULL; @@ -238,7 +233,8 @@ DEFINE_NATIVE_ENTRY(Isolate_spawnUri, 1) { ThrowIsolateSpawnException(msg); } - return Spawn(isolate, new IsolateSpawnState(canonical_uri)); + return Spawn(isolate, new IsolateSpawnState(port.Id(), canonical_uri, + args, message)); } @@ -259,14 +255,4 @@ DEFINE_NATIVE_ENTRY(Isolate_sendOOB, 2) { return Object::null(); } - -DEFINE_NATIVE_ENTRY(Isolate_mainPort, 0) { - // The control port is being accessed as a regular port from Dart code. This - // is most likely due to the _startIsolate code in dart:isolate. Account for - // this by increasing the number of open control ports. - isolate->message_handler()->increment_control_ports(); - - return ReceivePort::New(isolate->main_port()); -} - } // namespace dart diff --git a/runtime/lib/isolate_patch.dart b/runtime/lib/isolate_patch.dart index 00b26e07935..82e306fd007 100644 --- a/runtime/lib/isolate_patch.dart +++ b/runtime/lib/isolate_patch.dart @@ -77,6 +77,14 @@ void _isolateScheduleImmediate(void callback()) { _pendingImmediateCallback = callback; } +void _runPendingImmediateCallback() { + if (_pendingImmediateCallback != null) { + var callback = _pendingImmediateCallback; + _pendingImmediateCallback = null; + callback(); + } +} + /// The embedder can execute this function to get hold of /// [_isolateScheduleImmediate] above. Function _getIsolateScheduleImmediateClosure() { @@ -120,11 +128,7 @@ class _RawReceivePortImpl implements RawReceivePort { // VM. Once we have non-fatal global exceptions we need to catch errors // so that we can run the immediate callbacks. handler(message); - if (_pendingImmediateCallback != null) { - var callback = _pendingImmediateCallback; - _pendingImmediateCallback = null; - callback(); - } + _runPendingImmediateCallback(); } // Call into the VM to close the VM maintained mappings. @@ -181,69 +185,71 @@ typedef _MainFunctionArgsMessage(args, message); * * The initial startup message is received through the control port. */ -void _startIsolate(Function entryPoint, bool isSpawnUri) { - // This port keeps the isolate alive until the initial startup message has - // been received. - var keepAlivePort = new RawReceivePort(); - - ignoreHandler(message) { - // Messages on the current Isolate's control port are dropped after the - // initial startup message has been received. +void _startIsolate(SendPort parentPort, + Function entryPoint, + List args, + var message, + bool isSpawnUri, + RawReceivePort controlPort, + List capabilities) { + if (controlPort != null) { + controlPort.handler = (_) {}; // Nobody home on the control port. } + if (parentPort != null) { + // Build a message to our parent isolate providing access to the + // current isolate's control port and capabilities. + // + // TODO(floitsch): Send an error message if we can't find the entry point. + var readyMessage = new List(2); + readyMessage[0] = controlPort.sendPort; + readyMessage[1] = capabilities; - isolateStartHandler(message) { - // We received the initial startup message. Ignore all further messages and - // close the port which kept this isolate alive. - Isolate._self.handler = ignoreHandler; - keepAlivePort.close(); + // Out of an excess of paranoia we clear the capabilities from the + // stack. Not really necessary. + capabilities = null; + parentPort.send(readyMessage); + } + assert(capabilities == null); - SendPort replyTo = message[0]; - if (replyTo != null) { - // TODO(floitsch): don't send ok-message if we can't find the entry point. - replyTo.send("started"); - } - if (isSpawnUri) { - assert(message.length == 3); - List args = message[1]; - var isolateMessage = message[2]; - if (entryPoint is _MainFunctionArgsMessage) { - entryPoint(args, isolateMessage); - } else if (entryPoint is _MainFunctionArgs) { - entryPoint(args); - } else { - entryPoint(); - } + if (isSpawnUri) { + if (entryPoint is _MainFunctionArgsMessage) { + entryPoint(args, message); + } else if (entryPoint is _MainFunctionArgs) { + entryPoint(args); } else { - assert(message.length == 2); - var entryMessage = message[1]; - entryPoint(entryMessage); + entryPoint(); } + } else { + entryPoint(message); } - - Isolate._self.handler = isolateStartHandler; + _runPendingImmediateCallback(); } patch class Isolate { /* patch */ static Future spawn( void entryPoint(message), var message, { bool paused: false }) { // `paused` isn't handled yet. + RawReceivePort readyPort; try { // The VM will invoke [_startIsolate] with entryPoint as argument. - List spawnData = _spawnFunction(entryPoint); - assert(spawnData.length == 3); - SendPort controlPort = spawnData[0]; - RawReceivePort readyPort = new RawReceivePort(); - controlPort.send([readyPort.sendPort, message]); + readyPort = new RawReceivePort(); + _spawnFunction(readyPort.sendPort, entryPoint, message); Completer completer = new Completer.sync(); readyPort.handler = (readyMessage) { - assert(readyMessage == 'started'); readyPort.close(); + assert(readyMessage is List); + assert(readyMessage.length == 2); + SendPort controlPort = readyMessage[0]; + List capabilities = readyMessage[1]; completer.complete(new Isolate(controlPort, - pauseCapability: spawnData[1], - terminateCapability: spawnData[2])); + pauseCapability: capabilities[0], + terminateCapability: capabilities[1])); }; return completer.future; } catch (e, st) { + if (readyPort != null) { + readyPort.close(); + } return new Future.error(e, st); }; } @@ -251,41 +257,45 @@ patch class Isolate { /* patch */ static Future spawnUri( Uri uri, List args, var message, { bool paused: false }) { // `paused` isn't handled yet. + RawReceivePort readyPort; try { // The VM will invoke [_startIsolate] and not `main`. - List spawnData = _spawnUri(uri.toString()); - assert(spawnData.length == 3); - SendPort controlPort = spawnData[0]; - RawReceivePort readyPort = new RawReceivePort(); - controlPort.send([readyPort.sendPort, args, message]); + readyPort = new RawReceivePort(); + _spawnUri(readyPort.sendPort, uri.toString(), args, message); Completer completer = new Completer.sync(); readyPort.handler = (readyMessage) { - assert(readyMessage == 'started'); readyPort.close(); + assert(readyMessage is List); + assert(readyMessage.length == 2); + SendPort controlPort = readyMessage[0]; + List capabilities = readyMessage[1]; completer.complete(new Isolate(controlPort, - pauseCapability: spawnData[1], - terminateCapability: spawnData[2])); + pauseCapability: capabilities[0], + terminateCapability: capabilities[1])); }; return completer.future; } catch (e, st) { + if (readyPort != null) { + readyPort.close(); + } return new Future.error(e, st); }; return completer.future; } - static final RawReceivePort _self = _mainPort; - static RawReceivePort get _mainPort native "Isolate_mainPort"; - // TODO(iposva): Cleanup to have only one definition. // These values need to be kept in sync with the class IsolateMessageHandler // in vm/isolate.cc. static const _PAUSE = 1; static const _RESUME = 2; - static List _spawnFunction(Function topLevelFunction) + static SendPort _spawnFunction(SendPort readyPort, Function topLevelFunction, + var message) native "Isolate_spawnFunction"; - static List _spawnUri(String uri) native "Isolate_spawnUri"; + static SendPort _spawnUri(SendPort readyPort, String uri, + List args, var message) + native "Isolate_spawnUri"; static void _sendOOB(port, msg) native "Isolate_sendOOB"; diff --git a/runtime/vm/bootstrap_natives.h b/runtime/vm/bootstrap_natives.h index 691171e548a..1c9acdbe1d1 100644 --- a/runtime/vm/bootstrap_natives.h +++ b/runtime/vm/bootstrap_natives.h @@ -286,10 +286,9 @@ namespace dart { V(Int32x4_setFlagZ, 2) \ V(Int32x4_setFlagW, 2) \ V(Int32x4_select, 3) \ - V(Isolate_mainPort, 0) \ - V(Isolate_spawnFunction, 1) \ - V(Isolate_spawnUri, 1) \ - V(Isolate_sendOOB, 2) \ + V(Isolate_spawnFunction, 3) \ + V(Isolate_spawnUri, 4) \ + V(Isolate_sendOOB, 2) \ V(Mirrors_evalInLibraryWithPrivateKey, 2) \ V(Mirrors_makeLocalClassMirror, 1) \ V(Mirrors_makeLocalTypeMirror, 1) \ diff --git a/runtime/vm/isolate.cc b/runtime/vm/isolate.cc index cf51a8c2419..7be084e0787 100644 --- a/runtime/vm/isolate.cc +++ b/runtime/vm/isolate.cc @@ -771,17 +771,28 @@ static bool RunIsolate(uword parameter) { func ^= result.raw(); func = func.ImplicitClosureFunction(); + const Array& capabilities = Array::Handle(Array::New(2)); + Capability& capability = Capability::Handle(); + capability = Capability::New(isolate->pause_capability()); + capabilities.SetAt(0, capability); + capability = Capability::New(isolate->terminate_capability()); + capabilities.SetAt(1, capability); + // Instead of directly invoking the entry point we call '_startIsolate' with - // the entry point as argument. The '_startIsolate' function will - // communicate with the spawner to receive the initial message before it - // executes the real entry point. + // the entry point as argument. // Since this function ("RunIsolate") is used for both Isolate.spawn and // Isolate.spawnUri we also send a boolean flag as argument so that the // "_startIsolate" function can act corresponding to how the isolate was // created. - const Array& args = Array::Handle(Array::New(2)); - args.SetAt(0, Instance::Handle(func.ImplicitStaticClosure())); - args.SetAt(1, is_spawn_uri ? Bool::True() : Bool::False()); + const Array& args = Array::Handle(Array::New(7)); + args.SetAt(0, SendPort::Handle(SendPort::New(state->parent_port()))); + args.SetAt(1, Instance::Handle(func.ImplicitStaticClosure())); + args.SetAt(2, Instance::Handle(state->BuildArgs())); + args.SetAt(3, Instance::Handle(state->BuildMessage())); + args.SetAt(4, is_spawn_uri ? Bool::True() : Bool::False()); + args.SetAt(5, ReceivePort::Handle( + ReceivePort::New(isolate->main_port(), true /* control port */))); + args.SetAt(6, capabilities); const Library& lib = Library::Handle(Library::IsolateLibrary()); const String& entry_name = String::Handle(String::New("_startIsolate")); @@ -1109,7 +1120,6 @@ void Isolate::PrintJSON(JSONStream* stream, bool ref) { jsobj.AddProperty("depth", (intptr_t)0); } jsobj.AddProperty("livePorts", message_handler()->live_ports()); - jsobj.AddProperty("controlPorts", message_handler()->control_ports()); jsobj.AddProperty("pauseOnExit", message_handler()->pause_on_exit()); // TODO(turnidge): Make the debugger support paused_on_start/exit. @@ -1304,13 +1314,50 @@ T* Isolate::AllocateReusableHandle() { } -IsolateSpawnState::IsolateSpawnState(const Function& func) +static uint8_t* allocator(uint8_t* ptr, intptr_t old_size, intptr_t new_size) { + void* new_ptr = realloc(reinterpret_cast(ptr), new_size); + return reinterpret_cast(new_ptr); +} + + +static void SerializeObject(const Instance& obj, + uint8_t** obj_data, + intptr_t* obj_len) { + MessageWriter writer(obj_data, &allocator); + writer.WriteMessage(obj); + *obj_len = writer.BytesWritten(); +} + + +static RawInstance* DeserializeObject(Isolate* isolate, + uint8_t* obj_data, + intptr_t obj_len) { + if (obj_data == NULL) { + return Instance::null(); + } + SnapshotReader reader(obj_data, obj_len, Snapshot::kMessage, isolate); + const Object& obj = Object::Handle(isolate, reader.ReadObject()); + ASSERT(!obj.IsError()); + Instance& instance = Instance::Handle(isolate); + instance ^= obj.raw(); // Can't use Instance::Cast because may be null. + return instance.raw(); +} + + +IsolateSpawnState::IsolateSpawnState(Dart_Port parent_port, + const Function& func, + const Instance& message) : isolate_(NULL), + parent_port_(parent_port), script_url_(NULL), library_url_(NULL), class_name_(NULL), function_name_(NULL), - exception_callback_name_(NULL) { + exception_callback_name_(NULL), + serialized_args_(NULL), + serialized_args_len_(0), + serialized_message_(NULL), + serialized_message_len_(0) { script_url_ = NULL; const Class& cls = Class::Handle(func.Owner()); const Library& lib = Library::Handle(cls.library()); @@ -1324,19 +1371,30 @@ IsolateSpawnState::IsolateSpawnState(const Function& func) class_name_ = strdup(class_name.ToCString()); } exception_callback_name_ = strdup("_unhandledExceptionCallback"); + SerializeObject(message, &serialized_message_, &serialized_message_len_); } -IsolateSpawnState::IsolateSpawnState(const char* script_url) +IsolateSpawnState::IsolateSpawnState(Dart_Port parent_port, + const char* script_url, + const Instance& args, + const Instance& message) : isolate_(NULL), + parent_port_(parent_port), library_url_(NULL), class_name_(NULL), function_name_(NULL), - exception_callback_name_(NULL) { + exception_callback_name_(NULL), + serialized_args_(NULL), + serialized_args_len_(0), + serialized_message_(NULL), + serialized_message_len_(0) { script_url_ = strdup(script_url); library_url_ = NULL; function_name_ = strdup("main"); exception_callback_name_ = strdup("_unhandledExceptionCallback"); + SerializeObject(args, &serialized_args_, &serialized_args_len_); + SerializeObject(message, &serialized_message_, &serialized_message_len_); } @@ -1346,6 +1404,8 @@ IsolateSpawnState::~IsolateSpawnState() { free(function_name_); free(class_name_); free(exception_callback_name_); + free(serialized_args_); + free(serialized_message_); } @@ -1402,6 +1462,17 @@ RawObject* IsolateSpawnState::ResolveFunction() { } +RawInstance* IsolateSpawnState::BuildArgs() { + return DeserializeObject(isolate_, serialized_args_, serialized_args_len_); +} + + +RawInstance* IsolateSpawnState::BuildMessage() { + return DeserializeObject(isolate_, + serialized_message_, serialized_message_len_); +} + + void IsolateSpawnState::Cleanup() { SwitchIsolateScope switch_scope(I); Dart::ShutdownIsolate(); diff --git a/runtime/vm/isolate.h b/runtime/vm/isolate.h index b94d127c23f..7c338937bf4 100644 --- a/runtime/vm/isolate.h +++ b/runtime/vm/isolate.h @@ -803,12 +803,19 @@ class SwitchIsolateScope { class IsolateSpawnState { public: - explicit IsolateSpawnState(const Function& func); - explicit IsolateSpawnState(const char* script_url); + IsolateSpawnState(Dart_Port parent_port, + const Function& func, + const Instance& message); + IsolateSpawnState(Dart_Port parent_port, + const char* script_url, + const Instance& args, + const Instance& message); ~IsolateSpawnState(); Isolate* isolate() const { return isolate_; } void set_isolate(Isolate* value) { isolate_ = value; } + + Dart_Port parent_port() const { return parent_port_; } char* script_url() const { return script_url_; } char* library_url() const { return library_url_; } char* class_name() const { return class_name_; } @@ -817,15 +824,22 @@ class IsolateSpawnState { bool is_spawn_uri() const { return library_url_ == NULL; } RawObject* ResolveFunction(); + RawInstance* BuildArgs(); + RawInstance* BuildMessage(); void Cleanup(); private: Isolate* isolate_; + Dart_Port parent_port_; char* script_url_; char* library_url_; char* class_name_; char* function_name_; char* exception_callback_name_; + uint8_t* serialized_args_; + intptr_t serialized_args_len_; + uint8_t* serialized_message_; + intptr_t serialized_message_len_; }; } // namespace dart diff --git a/runtime/vm/message_handler.cc b/runtime/vm/message_handler.cc index 8bc3eb9cfaf..e5990fbbcfa 100644 --- a/runtime/vm/message_handler.cc +++ b/runtime/vm/message_handler.cc @@ -34,7 +34,6 @@ class MessageHandlerTask : public ThreadPool::Task { MessageHandler::MessageHandler() : queue_(new MessageQueue()), oob_queue_(new MessageQueue()), - control_ports_(0), live_ports_(0), paused_(0), pause_on_start_(false), @@ -272,8 +271,8 @@ void MessageHandler::ClosePort(Dart_Port port) { OS::Print("[-] Closing port:\n" "\thandler: %s\n" "\tport: %" Pd64 "\n" - "\tports: control(%" Pd ") live(%" Pd ")\n", - name(), port, control_ports_, live_ports_); + "\tports: live(%" Pd ")\n", + name(), port, live_ports_); } } @@ -307,22 +306,4 @@ void MessageHandler::decrement_live_ports() { live_ports_--; } - -void MessageHandler::increment_control_ports() { - MonitorLocker ml(&monitor_); -#if defined(DEBUG) - CheckAccess(); -#endif - control_ports_++; -} - - -void MessageHandler::decrement_control_ports() { - MonitorLocker ml(&monitor_); -#if defined(DEBUG) - CheckAccess(); -#endif - control_ports_--; -} - } // namespace dart diff --git a/runtime/vm/message_handler.h b/runtime/vm/message_handler.h index 78da45d72e1..65dc32f53c5 100644 --- a/runtime/vm/message_handler.h +++ b/runtime/vm/message_handler.h @@ -54,24 +54,13 @@ class MessageHandler { // Returns true on success. bool HandleOOBMessages(); - // The number of opened control ports is determined whether the isolate has - // live ports. An isolate is considered not having any live ports if only - // control ports are open. - // Usually either 0 or 1. - void increment_control_ports(); - void decrement_control_ports(); - // A message handler tracks how many live ports it has. - bool HasLivePorts() const { return live_ports_ > control_ports_; } + bool HasLivePorts() const { return live_ports_ > 0; } intptr_t live_ports() const { return live_ports_; } - intptr_t control_ports() const { - return control_ports_; - } - bool paused() const { return paused_ > 0; } void increment_paused() { paused_++; } @@ -168,7 +157,6 @@ class MessageHandler { Monitor monitor_; // Protects all fields in MessageHandler. MessageQueue* queue_; MessageQueue* oob_queue_; - intptr_t control_ports_; // The number of open control ports usually 0 or 1. intptr_t live_ports_; // The number of open ports, including control ports. intptr_t paused_; // The number of pause messages received. bool pause_on_start_; diff --git a/runtime/vm/object.cc b/runtime/vm/object.cc index f79dc5898e4..4eaf3069284 100644 --- a/runtime/vm/object.cc +++ b/runtime/vm/object.cc @@ -18808,7 +18808,9 @@ void Capability::PrintJSONImpl(JSONStream* stream, bool ref) const { } -RawReceivePort* ReceivePort::New(Dart_Port id, Heap::Space space) { +RawReceivePort* ReceivePort::New(Dart_Port id, + bool is_control_port, + Heap::Space space) { Isolate* isolate = Isolate::Current(); const SendPort& send_port = SendPort::Handle(isolate, SendPort::New(id)); @@ -18821,7 +18823,11 @@ RawReceivePort* ReceivePort::New(Dart_Port id, Heap::Space space) { result ^= raw; result.raw_ptr()->send_port_ = send_port.raw(); } - PortMap::SetLive(id); + if (is_control_port) { + PortMap::SetPortState(id, PortMap::kControlPort); + } else { + PortMap::SetPortState(id, PortMap::kLivePort); + } return result.raw(); } diff --git a/runtime/vm/object.h b/runtime/vm/object.h index 76bac3d757a..b8b4915ace6 100644 --- a/runtime/vm/object.h +++ b/runtime/vm/object.h @@ -6915,7 +6915,9 @@ class ReceivePort : public Instance { static intptr_t InstanceSize() { return RoundedAllocationSize(sizeof(RawReceivePort)); } - static RawReceivePort* New(Dart_Port id, Heap::Space space = Heap::kNew); + static RawReceivePort* New(Dart_Port id, + bool is_control_port, + Heap::Space space = Heap::kNew); private: FINAL_HEAP_OBJECT_IMPLEMENTATION(ReceivePort, Instance); diff --git a/runtime/vm/port.cc b/runtime/vm/port.cc index be861952832..f39cba06f62 100644 --- a/runtime/vm/port.cc +++ b/runtime/vm/port.cc @@ -63,6 +63,21 @@ void PortMap::Rehash(intptr_t new_capacity) { } +const char* PortMap::PortStateString(PortState kind) { + switch (kind) { + case kNewPort: + return "new"; + case kLivePort: + return "live"; + case kControlPort: + return "control"; + default: + UNREACHABLE(); + return "UNKNOWN"; + } +} + + Dart_Port PortMap::AllocatePort() { const Dart_Port kMASK = 0x3fffffff; Dart_Port result = prng_->NextUInt32() & kMASK; @@ -79,16 +94,21 @@ Dart_Port PortMap::AllocatePort() { } -void PortMap::SetLive(Dart_Port port) { +void PortMap::SetPortState(Dart_Port port, PortState state) { MutexLocker ml(mutex_); intptr_t index = FindPort(port); ASSERT(index >= 0); - map_[index].live = true; - map_[index].handler->increment_live_ports(); + PortState old_state = map_[index].state; + ASSERT(old_state == kNewPort); + map_[index].state = state; + if (state == kLivePort) { + map_[index].handler->increment_live_ports(); + } if (FLAG_trace_isolates) { - OS::Print("[^] Live port: \n" + OS::Print("[^] Port (%s) -> (%s): \n" "\thandler: %s\n" "\tport: %" Pd64 "\n", + PortStateString(old_state), PortStateString(state), map_[index].handler->name(), port); } } @@ -117,7 +137,7 @@ Dart_Port PortMap::CreatePort(MessageHandler* handler) { Entry entry; entry.port = AllocatePort(); entry.handler = handler; - entry.live = false; + entry.state = kNewPort; // Search for the first unused slot. Make use of the knowledge that here is // currently no port with this id in the port map. @@ -179,7 +199,7 @@ bool PortMap::ClosePort(Dart_Port port) { // pending messages below. map_[index].port = 0; map_[index].handler = deleted_entry_; - if (map_[index].live) { + if (map_[index].state == kLivePort) { handler->decrement_live_ports(); } @@ -203,7 +223,7 @@ void PortMap::ClosePorts(MessageHandler* handler) { // Mark the slot as deleted. map_[i].port = 0; map_[i].handler = deleted_entry_; - if (map_[i].live) { + if (map_[i].state == kLivePort) { handler->decrement_live_ports(); } used_--; diff --git a/runtime/vm/port.h b/runtime/vm/port.h index 5a0f177402b..f62ac8bd6ee 100644 --- a/runtime/vm/port.h +++ b/runtime/vm/port.h @@ -20,12 +20,18 @@ class PortMapTestPeer; class PortMap: public AllStatic { public: + enum PortState { + kNewPort = 0, // a newly allocated port + kLivePort = 1, // a regular port (has a ReceivePort) + kControlPort = 2, // a special control port (has a ReceivePort) + }; + // Allocate a port for the provided handler and return its VM-global id. static Dart_Port CreatePort(MessageHandler* handler); // Indicates that a port has had a ReceivePort created for it at the // dart language level. The port remains live until it is closed. - static void SetLive(Dart_Port id); + static void SetPortState(Dart_Port id, PortState kind); // Close the port with id. All pending messages will be dropped. // @@ -59,9 +65,11 @@ class PortMap: public AllStatic { typedef struct { Dart_Port port; MessageHandler* handler; - bool live; + PortState state; } Entry; + static const char* PortStateString(PortState state); + // Allocate a new unique port. static Dart_Port AllocatePort(); diff --git a/runtime/vm/port_test.cc b/runtime/vm/port_test.cc index 64252dc3abf..34272d150b7 100644 --- a/runtime/vm/port_test.cc +++ b/runtime/vm/port_test.cc @@ -25,7 +25,7 @@ class PortMapTestPeer { if (index < 0) { return false; } - return PortMap::map_[index].live; + return PortMap::map_[index].state == PortMap::kLivePort; } }; @@ -100,20 +100,36 @@ TEST_CASE(PortMap_CreateManyPorts) { } -TEST_CASE(PortMap_SetLive) { +TEST_CASE(PortMap_SetPortState) { PortTestMessageHandler handler; + + // Regular port. Dart_Port port = PortMap::CreatePort(&handler); EXPECT_NE(0, port); EXPECT(PortMapTestPeer::IsActivePort(port)); EXPECT(!PortMapTestPeer::IsLivePort(port)); - PortMap::SetLive(port); + PortMap::SetPortState(port, PortMap::kLivePort); EXPECT(PortMapTestPeer::IsActivePort(port)); EXPECT(PortMapTestPeer::IsLivePort(port)); PortMap::ClosePort(port); EXPECT(!PortMapTestPeer::IsActivePort(port)); EXPECT(!PortMapTestPeer::IsLivePort(port)); + + // Control port. + port = PortMap::CreatePort(&handler); + EXPECT_NE(0, port); + EXPECT(PortMapTestPeer::IsActivePort(port)); + EXPECT(!PortMapTestPeer::IsLivePort(port)); + + PortMap::SetPortState(port, PortMap::kControlPort); + EXPECT(PortMapTestPeer::IsActivePort(port)); + EXPECT(!PortMapTestPeer::IsLivePort(port)); + + PortMap::ClosePort(port); + EXPECT(!PortMapTestPeer::IsActivePort(port)); + EXPECT(!PortMapTestPeer::IsLivePort(port)); }